| 123456789101112131415161718192021 |
- from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
- from app.config import settings
- # 创建异步数据库引擎
- engine = create_async_engine(settings.database_url, echo=False)
- # 异步 Session 工厂
- async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
- async def get_db():
- """FastAPI 依赖注入，提供数据库 Session"""
- async with async_session() as session:
- yield session
- async def create_tables():
- """开发环境：通过 SQLAlchemy 直接建表（生产环境请使用 SQL 迁移文件）"""
- from app.models.domain import Base
- async with engine.begin() as conn:
- await conn.run_sync(Base.metadata.create_all)
|
