---
description: "需要和用户确认使用哪种数据库以及是使用同步还是异步数据库"
---

# 异步数据库 (SQLModel)

需要和用户确认使用哪种数据库以及是使用同步还是异步数据库

## 依赖

只需在最后提醒需要安装什么，不要自己安装，由用户自行安装


```bash
# PostgreSQL
uv add sqlmodel asyncpg

# SQLite
uv add sqlmodel aiosqlite
```

## 创建异步引擎

```python
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 集成

```python
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 向量库）时，需要注意：

```python
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=` 参数指定具体表**，防止在错误的数据库中建表：

```python
# 异步建表 - 只建主业务表
async with engine.begin() as conn:
    await conn.run_sync(
        SQLModel.metadata.create_all,
        tables=[SessionTable.__table__, MessageTable.__table__]
    )

# 同步建表 - 只建向量库表
SQLModel.metadata.create_all(engine_faq, tables=[FAQ_CREATE.__table__])
```