如何为 FastAPI 应用程序编写 SQLAlchemy 测试夹具

Posted 2023-02-15

【中文标题】如何为 FastAPI 应用程序编写 SQLAlchemy 测试夹具

【英文标题】：How do I write SQLAlchemy test fixtures for FastAPI applications

【发布时间】：2020-06-27 21:30:45

【问题描述】：

我正在编写一个使用 SQLAlchemy 数据库的FastAPI 应用程序。我复制了example from the FastAPI documentation，为了简洁起见简化了数据库模式。完整的源代码在这篇文章的底部。

这行得通。我可以使用uvicorn sql_app.main:app 运行它，并通过 Swagger 文档与数据库交互。当它运行时，它会在工作目录中创建一个test.db。

现在我想添加一个单元测试。像这样。

from fastapi import status
from fastapi.testclient import TestClient
from pytest import fixture

from main import app


@fixture
def client() -> TestClient:
    return TestClient(app)


def test_fast_sql(client: TestClient):
    response = client.get("/users/")
    assert response.status_code == status.HTTP_200_OK
    assert response.json() == []

使用下面的源代码，这会将工作目录中的test.db 作为数据库。相反，我想为每个在测试结束时删除的单元测试创​​建一个新数据库。

我可以将全局 database.engine 和 database.SessionLocal 放在运行时创建的对象中，如下所示：

    class UserDatabase:
        def __init__(self, directory: Path):
            directory.mkdir(exist_ok=True, parents=True)
            sqlalchemy_database_url = f"sqlite:///directory/store.db"
            self.engine = create_engine(
                sqlalchemy_database_url, connect_args="check_same_thread": False
            )
            self.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=self.engine)
            models.Base.metadata.create_all(bind=self.engine)

但我不知道如何使用 main.get_db 进行这项工作，因为 Depends(get_db) 逻辑最终假定 database.engine 和 database.SessionLocal 在全球范围内可用。

我习惯使用 Flask，它的单元测试工具会为您处理所有这些。我自己不知道怎么写。有人能告诉我为了在这个框架中为每个单元测试生成一个新数据库，我必须做的最小更改吗？

---

简化的 FastAPI/SQLAlchemy 应用的完整源码如下。

database.py

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

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()

models.py

from sqlalchemy import Column, Integer, String

from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String)
    age = Column(Integer)

schemas.py

from pydantic import BaseModel


class UserBase(BaseModel):
    name: str
    age: int


class UserCreate(UserBase):
    pass


class User(UserBase):
    id: int

    class Config:
        orm_mode = True

crud.py

from sqlalchemy.orm import Session

import schemas
import models


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    db_user = models.User(name=user.name, age=user.age)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

ma​​in.py

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

import schemas
import models
import crud
from database import SessionLocal, engine

models.Base.metadata.create_all(bind=engine)

app = FastAPI()


# Dependency
def get_db():
    try:
        db = SessionLocal()
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/user_id", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

【问题讨论】：

FastAPI 网站上的相关问题：github.com/tiangolo/fastapi/issues/831

【参考方案1】：

您需要在测试中覆盖 get_db 依赖项，请参阅 these docs。

你的灯具是这样的：

@fixture
def db_fixture() -> Session:
    raise NotImplementError()  # Make this return your temporary session

@fixture
def client(db_fixture) -> TestClient:

    def _get_db_override():
        return db_fixture

    app.dependency_overrides[get_db] = _get_db_override
    return TestClient(app)

【讨论】：

实际上，这里似乎直接解决了这个问题：fastapi.tiangolo.com/advanced/testing-database。我忽略了那个文档。