异步数据库 (SQLModel)
需要和用户确认使用哪种数据库以及是使用同步还是异步数据库
依赖
只需在最后提醒需要安装什么，不要自己安装，由用户自行安装
创建异步引擎
from sqlalchemy.ext.asyncio import create_async_engine
from sqlmodel.ext.asyncio.session import AsyncSession
# PostgreSQL
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
async_engine = create_async_engine(DATABASE_URL, echo=True)
# SQLite
DATABASE_URL = "sqlite+aiosqlite:///./database.db"
async_engine = create_async_engine(
DATABASE_URL,
echo=True,
connect_args={"check_same_thread": False}, # SQLite 必需
)
async def get_session():
async with AsyncSession(async_engine, expire_on_commit=False) as session:
yield session
与 lifespan 集成
from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
@asynccontextmanager
async def lifespan(app: FastAPI):
# 异步建表
async with async_engine.begin() as conn:
await conn.run_sync(SQLModel.metadata.create_all)
yield
await async_engine.dispose()
app = FastAPI(lifespan=lifespan)
# 使用依赖注入
@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
user = await session.get(User, user_id)
return user
同步数据库 (SQLModel)
同步数据库已经比较成熟，自由发挥即可
同步 + 异步双引擎模式
当项目需要同时使用异步引擎（主业务）和同步引擎（如 sqlite-vec 向量库）时，需要注意：
from sqlalchemy.ext.asyncio import create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlmodel.ext.asyncio.session import AsyncSession
from sqlmodel import create_engine, Session
# 异步引擎 - 主业务数据库
engine = create_async_engine("sqlite+aiosqlite:///volume/dbs/chat.db", echo=False)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# 同步引擎 - 向量库等不支持异步的场景
engine_faq = create_engine("sqlite:///volume/dbs/faq.db")
# 分别注入
async def get_session() -> AsyncGenerator[AsyncSession, None]:
async with async_session(expire_on_commit=False) as session:
yield session
def get_session_faq() -> Session:
with Session(engine_faq) as session:
yield session
建表时必须用 tables= 参数指定具体表，防止在错误的数据库中建表：