如何为SQLAlchemy应用程序编写FastAPI测试夹具
如何为SQLAlchemy应用程序编写FastAPI测试夹具
提问于 2020-03-16 18:42:24
我正在编写一个使用FastAPI数据库的SQLAlchemy应用程序。我复制了FastAPI文档中的示例，简化了数据库模式。完整的资料来源在这篇文章的底部。
这个很管用。我可以使用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 = Truecrud.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_usermain.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回答 1
Stack Overflow用户
回答已采纳
发布于 2020-04-05 15:56:00
您需要在测试中覆盖您的get_db依赖项，请参阅这些医生。
就像你的夹具一样：
@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)页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/60711576
复制相关文章
相似问题
腾讯云开发者
