fastapi/backend/app/extensions/db.py

from app.config import settings
from app.models import crud
from app.models.user import User, UserCreate
from sqlmodel import Session, SQLModel, create_engine, select

# postgres
engine = create_engine(str(settings.SQLALCHEMY_DATABASE_URI))


async def initiate_nongo():
    client = AsyncIOMotorClient(Settings().DATABASE_URL)
    await init_beanie(
        database=client.get_default_database(), document_models=models.__all__
    )


async def initate_postgres(session: Session) -> None:
    """
    init db
    :param session: database session
    """
    # Base.metadata.create_all(bind=engine)
    SQLModel.metadata.create_all(bind=engine)
    user = session.exec(
        select(User).where(User.email == settings.FIRST_SUPERUSER)
    ).first()
    # 插入管理员账户
    if not user:
        user_in = UserCreate(
            email=settings.FIRST_SUPERUSER,
            password=settings.FIRST_SUPERUSER_PASSWORD,
            is_superuser=True,
        )
        user = crud.create_user(session=session, user_create=user_in)


async def initate_redis():
    await init_redis(app=app, settings=settings)


async def initate_sqlite():
    await init_sqlite(app=app, settings=settings)