使用FastAPI和PostgreSQL进行测试[已关闭]

7cjasjjr  于 2023-06-29  发布在  PostgreSQL
关注(0)|答案(1)|浏览(154)

已关闭,此问题为opinion-based。目前不接受答复。
**想改善这个问题吗?**更新问题,以便editing this post可以用事实和引用来回答。

6天前关闭
Improve this question
我正在开发一个连接到PostgreSQL数据库的FastAPI后端应用程序,我有点被测试的良好实践卡住了。我读了很多stackoverflow postsblogs,但我并不真正习惯后端开发,我仍然不真正理解什么是最佳实践。
知道我使用SQLModel后,文档中建议使用内存中的DBSQLite执行测试。问题是,当我遵循所解释的方法时,PG和SQLite(关于模式)之间的不兼容性令我震惊。关键是我使用的是一个具有多个模式的现有DB,而不仅仅是一个公共模式。因此,当我运行测试时,我遇到错误“schema pouetpouet doesn’t exist”。
最后,问题是:我应该怎么做来测试我的应用?
1.找到一种方法来设置我的prod Postgres DB和内存中的SQLite DB之间的兼容性?
1.在preprod Postgres DB上应用我的测试,并尝试清理添加/删除的项目?(我实际上做了什么,但我不认为这是一个真正的好做法)
1.在Docker容器中设置本地Postgres服务器?
1.在pytest测试文件中用Dict模拟一个数据库?
1.使用第三个库,例如testcontainers
1.不做检查?
毕竟,我喜欢做单元测试和集成测试,所以也许我的需求不止一个解决方案。
以下是我的项目的简化版本:
我的项目架构:(假设每个文件夹中都有一个__ init__.py文件)

app/
├── api/
│   ├── core/
│   │   ├── config.py   #get the env settings and distribute it to the app
│   │   ├── .env
│   ├── crud/
│   │   ├── items.py    #the CRUD functions called by the router
│   ├── db/
│   │   ├── session.py  #the get_session function handling the db engine
│   ├── models/
│   │   ├── items.py    #the SQLModel object def as is in the db
│   ├── routers/
│   │   ├── items.py    #the routing system
│   ├── schemas/
│   │   ├── items.py    #the python object def as it is used in the app
│   ├── main.py         #the main app
├── tests/
│   ├── test_items.py   #the pytest testing file

crud/items.py中:

from fastapi.encoders import jsonable_encoder
from sqlmodel import Session, select
from api.models import Item
from api.schemas import ItemCreate

def get_item(db_session: Session, item_id: int) -> Item:
    query = select(Item).where(Item.id == item_id)
    return db_session.exec(query).first()

def create_new_item(db_session: Session, *, obj_input: ItemCreate) -> Item:
    obj_in_data = jsonable_encoder(obj_input)
    db_obj = Item(**obj_in_data)
    db_session.add(db_obj)
    db_session.commit()
    db_session.refresh(db_obj)
    return db_obj

db/session.py中:

from sqlalchemy.engine import Engine
from sqlmodel import create_engine, Session
from api.core.config import settings

engine: Engine = create_engine(settings.SQLALCHEMY_DATABASE_URI, pool_pre_ping=True)

def get_session() -> Session:
    with Session(engine) as session:
        yield session

models/items.py中:

from sqlmodel import SQLModel, Field, MetaData

meta = MetaData(schema="pouetpouet")  # https://github.com/tiangolo/sqlmodel/issues/20

class Item(SQLModel, table=True):
    __tablename__ = "cities"
    # __table_args__ = {"schema": "pouetpouet"}
    metadata = meta

    id: int = Field(primary_key=True, default=None)
    city_name: str

routers/items.py中:

from fastapi import APIRouter, Depends, HTTPException
from sqlmodel import Session
from api.crud import get_item, create_new_item
from api.db.session import get_session
from api.models import Item
from api.schemas import ItemRead, ItemCreate

router = APIRouter(prefix="/api/items", tags=["Items"])

@router.get("/{item_id}", response_model=ItemRead)
def read_item(
    *,
    db_session: Session = Depends(get_session),
    item_id: int,
) -> Item:
    item = get_item(db_session=db_session, item_id=item_id)
    if not item:
        raise HTTPException(status_code=404, detail="Item not found")
    return item

@router.post("/", response_model=ItemRead)
def create_item(
    *,
    db_session: Session = Depends(get_session),
    item_input: ItemCreate,
) -> Item:
    item = create_new_item(db_session=db_session, obj_input=item_input)
    return item

schemas/items.py中:

from typing import Optional
from sqlmodel import SQLModel

class ItemBase(SQLModel):
    city_name: Optional[str] = None

class ItemCreate(ItemBase):
    pass

class ItemRead(ItemBase):
    id: int
    class Config:
        orm_mode: True

tests/test_items.py中:

from fastapi.testclient import TestClient
from api.main import app

client = TestClient(app)

def test_create_item() -> None:
    data = {"city_name": "Las Vegas"}
    response = client.post("/api/items/", json=data)
    assert response.status_code == 200
    content = response.json()
    assert content["city_name"] == data["city_name"]
    assert "id" in content

附言:在后端开发方面不是很有经验,如果你注意到一些奇怪的东西,不要犹豫,对我的代码提出建设性的意见。它会受到很好的欢迎。

vx6bjr1n

vx6bjr1n1#

当你开始测试时,测试总是一件令人困惑的事情,有很多观点和一些一成不变的东西。
也就是说,你的提问方法似乎是正确的,而且经过了深思熟虑,所以我将尝试回答(请记住,这来自个人偏好和到目前为止为众多应用程序编写测试的经验,所以它将主要是固执己见):
1.我不会这样做,因为这可能是太多的工作,并导致非常脆弱的测试。如果你解决了兼容性问题,如果/当事情破裂时,你将很难找出真实的的问题是什么。
1.这是一个有效的方法:您可以实现一组具有可预测结果的集成测试,在Staging环境中部署代码并针对该环境运行套件。您甚至可以在CI/CD管道的一个步骤中运行这些测试以实现自动化。
1.* 这是本地开发的有效方法 *:在本地开发和测试新的实现时使用它。你可以在单元测试中使用它,但是你必须在每次运行后进行清理(可能需要编写脚本来自动清理),以避免残留和测试失败或意外通过。
1.这是一种有效的方法,但对于测试数据库来说效果不太好。我通常使用它来测试API和从DB获取的数据的“后处理”方法。换句话说,仅仅因为模拟有效,并不意味着最终结果也有效。读取interesting article about this btw
1.这是一个有效的方法testcontainers提供了一个容器化的数据库,可以生成测试,我们可以毫无顾虑地向其中添加模式和表(如果这里有什么问题,意味着测试提前发现了问题),然后它就会被取消。我会用这种方法来解决这个问题!(here is a basic example of using it with pytest
1.恐惧导致不使用测试,不使用测试导致许多不眠之夜,许多不眠之夜导致更多的问题。这是通往黑暗面的道路(以及许多失去的日子,甚至周末)。
最后,请记住,这里没有“银”,因此可以一起使用不同的方法。
例如,在您的案例中,我会使用testcontainers进行本地和预提交测试,然后在部署到PROD之前在Staging环境中运行集成测试套件。
您的代码处理方法看起来不错。
这里有一点小小的评论:由于您使用的是SQLModel,因此可以避免使用单独的SQLAlchemy models和Pydantic schemas,它们用于SQLAlchemy的“传统”FastAPI方法。您可以直接将SQLModel用作模式,而不是as demonstrated in the documentation
希望这能帮助你找到一个好的解决方案,为您的测试:)

相关问题