FastAPI中基于Repository模式封装SQL会话及实现可测试性的方法问询

我明白你想把SQLModel的数据库操作封装到Repository类里，让接口层完全依赖Repository，同时还要保证每个请求有独立的数据库会话，并且能轻松做分层测试——这确实是FastAPI里实现整洁架构的常见需求，我来一步步给你捋清楚怎么实现：

一、调整Repository实现，让它成为可注入的依赖 #

你原来的写法把SessionDep放到了Repository的方法参数里，这样接口调用的时候还要手动传会话，不符合依赖注入的思路。咱们可以把Session作为Repository的初始化参数，再把Repository本身做成一个FastAPI依赖，这样每个请求会自动拿到绑定当前会话的Repository实例：

1. 先保留数据库会话的基础依赖 #

这部分和你原来的代码差不多，不用大改：

from sqlmodel import create_engine, Session, select
from fastapi import Depends, APIRouter
from typing import Annotated, Generator

SQLITE_FILE_PATH = "test.db"
engine = create_engine(f"sqlite:///{SQLITE_FILE_PATH}", connect_args={"check_same_thread": False})

def get_session() -> Generator[Session, None, None]:
    with Session(engine) as session:
        yield session

SessionDep = Annotated[Session, Depends(get_session)]

2. 重构Repository类，通过构造方法注入会话 #

把会话存在Repository的实例属性里，这样所有方法都能直接用，不用每次传参数：

# 假设你的Resource模型是这样定义的（如果不是，替换成你自己的即可）
from sqlmodel import SQLModel, Field
class Resource(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str

class ResourceRepository:
    def __init__(self, session: Session):
        self.session = session

    def list(self) -> list[Resource]:
        query = select(Resource)
        # 注意这里要加.all()，不然返回的是Result对象，无法直接序列化为接口响应
        resources = self.session.exec(query).all()
        return resources

3. 把Repository封装成FastAPI依赖 #

写一个依赖函数，每次请求时用当前的Session实例创建Repository：

def get_resource_repo(session: SessionDep) -> ResourceRepository:
    return ResourceRepository(session)

# 定义Repository的依赖注解，方便接口里使用
RepoDep = Annotated[ResourceRepository, Depends(get_resource_repo)]

4. 接口层直接依赖Repository #

现在接口里只需要注入RepoDep，不用管数据库会话的细节了：

router = APIRouter()

@router.get("/", response_model=list[Resource])
def list_resources(repo: RepoDep):
    return repo.list()

这样做的好处是：每个请求会通过get_session拿到独立的会话，再通过get_resource_repo创建绑定该会话的Repository实例——既保证了每个请求的会话隔离，又让接口层完全和数据库细节解耦。

二、实现分层测试（接口层+Repository层） #

分层测试的核心是“隔离”：接口层测试mock掉Repository，只验证接口逻辑；Repository层测试用内存数据库，只验证数据访问逻辑。

1. 接口层测试：Mock Repository依赖 #

用pytest-mock配合FastAPI的dependency_overrides，把真实的Repository替换成mock对象，这样不用启动真实数据库就能测试接口：

from fastapi.testclient import TestClient
from your_main_file import app, get_resource_repo, ResourceRepository

client = TestClient(app)

def test_list_endpoint_mock_repo(mocker):
    # 创建mock的Repository实例，预设list方法的返回值
    mock_repo = mocker.Mock(spec=ResourceRepository)
    mock_repo.list.return_value = [Resource(id=1, name="test_resource")]

    # 覆盖依赖：让接口请求时拿到的是我们的mock实例
    app.dependency_overrides[get_resource_repo] = lambda: mock_repo

    # 调用接口
    response = client.get("/")
    assert response.status_code == 200
    # 验证响应内容符合预期
    assert response.json() == [{"id": 1, "name": "test_resource"}]

    # 验证mock的list方法确实被调用了，确保接口逻辑正确
    mock_repo.list.assert_called_once()

    # 测试完记得清空依赖覆盖，避免影响其他测试
    app.dependency_overrides.clear()

2. Repository层测试：用内存SQLite数据库 #

用内存数据库做测试，既快又不会污染真实数据，还能完全模拟真实的数据库操作：

import pytest
from sqlmodel import SQLModel
from your_main_file import ResourceRepository, Resource

@pytest.fixture(name="test_session")
def fixture_test_session():
    # 创建内存SQLite引擎，connect_args是SQLite的特殊参数，避免线程问题
    test_engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
    # 创建所有表结构
    SQLModel.metadata.create_all(test_engine)
    # 提供会话给测试用例
    with Session(test_engine) as session:
        yield session

def test_repo_list(test_session):
    # 先往测试数据库里插入一条测试数据
    test_resource = Resource(name="test_data")
    test_session.add(test_resource)
    test_session.commit()
    test_session.refresh(test_resource)

    # 创建Repository实例，传入测试会话
    repo = ResourceRepository(test_session)
    # 调用要测试的方法
    result = repo.list()

    # 验证结果是否符合预期
    assert len(result) == 1
    assert result[0].name == "test_data"
    assert result[0].id == test_resource.id

三、额外注意点 #

1. 如果你的Repository有多个方法（比如create、update、delete），都可以直接用self.session来操作，不用重复传参数；
2. 接口的response_model要和Repository返回的类型匹配，比如这里用list[Resource]，FastAPI会自动帮你做序列化；
3. 测试完记得清空dependency_overrides，不然会影响其他测试用例的依赖注入。

备注：内容来源于stack exchange，提问作者fedest