FastAPI中非装饰函数调用SQL database.exec()报错的问题求助

我正在优化团队的FastAPI接口，需求是仅当条目不存在时，将其注册到SQL数据库。

数据库会话定义如下：

from functools import partial

from sqlmodel import Session, create_engine
from fastapi import Depends

def get_db_session() -> Session:  # type: ignore
    _url = "postgresql+psycopg2://some-path:8000"
    engine = create_engine(_url)
    with Session(engine) as session:
        try:
            yield session
        finally:
            session.close()

our_db_session = partial(get_db_session, some_config_file)
our_db: Session = Depends(our_db_session)

初始实现中，查询逻辑放在路由函数内时运行正常：

from fastapi import APIRouter

from other_file import our_db

router = APIRouter()

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name==file_name)
        .where(Table1.some_value==search_criteria)
    ).one()

为了减少重复代码，我将查询逻辑抽成独立函数并在路由内调用，此时报错："Depends没有属性.exec()"

def get_db_entry(
    file_name: Path,
    search_criteria: Union[str, int],
    db: Session = our_db,
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name==file_name)
        .where(Table1.some_value==search_criteria)
    ).one()
    return db_entry

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    db_entry = get_db_entry(file_name=file_name, search_criteria=search_criteria)

FastAPI的依赖注入机制**仅会自动解析路由处理函数（被@router.post等装饰器修饰的函数）**参数中的Depends对象，将其替换为实际的依赖实例（这里就是SQLModel的Session对象）。

而独立函数get_db_entry不属于路由处理函数，FastAPI不会自动解析它的参数默认值。当你调用get_db_entry时未传入db参数，它拿到的是our_db对应的Depends对象本身，而非实际的Session实例，所以会报错Depends没有属性.exec()。

方案一：手动传入已解析的Session实例（推荐） #

修改独立函数，去掉db参数的默认值，明确要求传入Session实例，然后在路由函数中把已经被FastAPI解析好的db传给它：

def get_db_entry(
    file_name: Path,
    search_criteria: Union[str, int],
    db: Session,  # 移除默认值，强制传入Session实例
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name == file_name)
        .where(Table1.some_value == search_criteria)
    ).one()
    return db_entry

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    db: Session = our_db,
):
    # 将已解析的db实例传入get_db_entry
    db_entry = get_db_entry(file_name=file_name, search_criteria=search_criteria, db=db)

方案二：将查询函数改为依赖注入 #

如果需要在多个路由中复用该查询结果，可以把get_db_entry改成依赖函数，让FastAPI自动解析其内部的Depends并返回查询结果：

from fastapi import Depends

def get_db_entry(
    file_name: Path,
    search_criteria: Union[str, int],
    db: Session = our_db,  # 作为依赖函数时，FastAPI会自动解析这里的Depends
):
    db_entry = db.exec(
        select(Table1)
        .where(Table1.file_name == file_name)
        .where(Table1.some_value == search_criteria)
    ).one()
    return db_entry

@router.post("/register_file/")
def register_file(
    file_name: Path,
    search_criteria: Union[int, str],
    # 直接注入查询结果
    db_entry: Table1 = Depends(get_db_entry),
):
    # 直接使用db_entry即可
    pass

内容的提问来源于stack exchange，提问作者TheEponymousProgrammer