如何让SQLAlchemy模型像Django模型一样在FastAPI中使用？

可以通过上下文变量存储会话或者基类封装会话获取逻辑来避免频繁传递db会话参数，以下是具体实现方案：

方案1：上下文变量+模型类封装（最接近Django体验） #

1. 配置上下文变量与会话依赖 #

先定义上下文变量存储当前请求的会话，同时修改默认的get_db依赖，将会话存入上下文：

import contextvars
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# 数据库基础配置
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

# 上下文变量，用于存储当前请求的db会话
db_session_var = contextvars.ContextVar("db_session")

def get_db():
    db = SessionLocal()
    token = db_session_var.set(db)
    try:
        yield db
    finally:
        db_session_var.reset(token)
        db.close()

2. 封装模型类的查询入口 #

创建抽象基模型类，模仿Django的objects管理器：

from sqlalchemy import Column, Integer, String

class BaseModel(Base):
    __abstract__ = True  # 标记为抽象类，不会生成数据库表

    @classmethod
    def objects(cls):
        """获取当前模型的查询对象，用法和Django的SomeModel.objects一致"""
        try:
            db = db_session_var.get()
            return db.query(cls)
        except LookupError:
            raise RuntimeError("必须在FastAPI请求上下文内调用该方法，请确保路由依赖了get_db")

# 业务模型示例
class SomeModel(BaseModel):
    __tablename__ = "some_models"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), index=True)

3. 实际使用 #

路由中只需依赖get_db，无需手动传递会话给模型查询：

from fastapi import Depends, FastAPI

app = FastAPI()

@app.get("/some-models/")
def list_some_models(db: Session = Depends(get_db)):
    # 完全模仿Django的查询写法
    return SomeModel.objects().filter(SomeModel.name.contains("test")).all()

方案2：基仓库类封装会话 #

如果习惯使用仓库模式，可以创建基仓库类自动获取会话：

class BaseRepository:
    @property
    def db(self) -> Session:
        try:
            return db_session_var.get()
        except LookupError:
            raise RuntimeError("仓库方法必须在FastAPI请求上下文内调用")

class SomeRepository(BaseRepository):
    def get_all_items(self):
        # 直接使用self.db，无需在方法参数中传递会话
        return self.db.query(SomeModel).all()

# 路由中使用仓库
@app.get("/items/")
def read_items(db: Session = Depends(get_db)):
    repo = SomeRepository()
    return repo.get_all_items()

注意事项 #

• 所有查询操作必须在FastAPI请求上下文中执行，即路由必须依赖get_db，否则上下文变量中没有会话会抛出错误。
• 如果使用异步FastAPI（配合async-sqlalchemy），上下文变量方案依然适用，只需将同步会话替换为异步会话即可。
• 后台任务、定时任务等请求上下文之外的场景，需要手动创建并传入会话。

内容的提问来源于stack exchange，提问作者Antonio Gamiz Delgado