FastAPI 和 PostgreSQL 的测试

huangapple go评论87阅读模式
英文:

Tests with FastAPI and PostgreSQL

问题

我正在使用FastAPI开发一个与PostgreSQL数据库连接的后端应用程序,但在测试方面遇到了一些问题。我阅读了很多关于此问题的stackoverflow帖子1和博客2,但我对后端开发不太熟悉,仍然不太理解什么是最佳实践。

考虑到我使用了SQLModel,文档中建议3使用内存中的SQLite数据库执行测试。问题是,当我按照文档中的方法操作时,PG和SQLite之间存在不兼容性(关于模式)。问题在于,我正在使用具有多个模式而不仅仅是公共模式的现有数据库。因此,当我运行我的测试时,我遇到了错误"schema pouetpouet不存在"。

最后,我的问题是:我应该如何测试我的应用程序?

  1. 找到一种设置我的生产Postgres数据库与内存中SQLite数据库兼容的方法吗?
  2. 在预生产Postgres数据库上运行测试,然后尝试清理已添加/删除的项目吗?(实际上是我做的,但我认为这不是一个很好的实践)
  3. 在Docker容器中设置本地Postgres服务器?
  4. 在pytest测试文件中使用类似字典的方式模拟数据库?
  5. 使用第三方库,例如testcontainers吗?
  6. 不进行测试?

最终,我想进行单元测试和集成测试,所以也许关于我的需求不止一种解决方案。

你的项目架构如下:

我的项目架构如下(考虑到每个文件夹中都有一个__ init__.py文件):

app/
├── api/
│   ├── core/
│   │   ├── config.py   #获取环境设置并将其分发给应用程序
│   │   ├── .env
│   ├── crud/
│   │   ├── items.py    #由路由器调用的CRUD函数
│   ├── db/
│   │   ├── session.py  #处理数据库引擎的get_session函数
│   ├── models/
│   │   ├── items.py    #作为数据库中的SQLModel对象定义
│   ├── routers/
│   │   ├── items.py    #路由系统
│   ├── schemas/
│   │   ├── items.py    #作为应用程序中使用的Python对象定义
│   ├── main.py         #主应用程序
├── tests/
│   ├── test_items.py   #pytest测试文件

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"
    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

请注意,我在后端开发方面经验有限,如果你发现代码中有什么奇怪的地方,欢迎提出建设性意见。

英文:

I'm developing a backend app with FastAPI connected to a PostgreSQL database and I'm a bit stuck and lost with the tests good practices.
I read a lot stackoverflow posts and blogs but I'm not really used to backend development and I still doesn't really understand what is the best practice.

Knowing that I use SQLModel, it is suggested in the documentation to perform the tests with a DB SQLite in memory. The problem is that when I follow the explained approach, I am struck by the non-compatibility between PG and SQLite (about schemas). The point is that I am consuming an existing DB with several schemas and not just a public schema. So, when I run my tests, I encounter the error "schema pouetpouet does not exist".

Finally, the question is: What should I do to test my app ?

  1. Find a way to setup the compatibility between my prod Postgres DB and an in-memory SQLite DB ?
  2. Apply my tests on a preprod Postgres DB and try to cleanup the added/removed items ? (what I did actually but I don't think it is a really good practice)
  3. Setup a local Postgres server inside a Docker container ?
  4. Mock a DB with kind of a Dict in the pytest test file ?
  5. Use a third lib like testcontainers for exemple ?
  6. Don't do tests ?

After all, I'd like to do unit and integration tests so maybe there is not only one solution about my needs.


Here is a really simplified version of my project:

The architecture of my project: (Consider that there is an __ init__.py file in each folder)

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

In the 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

In the 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

In the 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

In the 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

In the 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

In the 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

ps: not being very experienced in backend development, do not hesitate to bring constructive remarks about my code if you notice something strange. It will be very well received.

答案1

得分: 3

测试始终是一个令人困惑的问题,当你开始进行测试时,会有很多不同的观点,但很少有事情是确定的。

话虽如此,你的质疑方法似乎是正确的,经过深思熟虑,所以我会尝试回答(请记住,这仅来自个人偏好和在多种应用程序上编写测试的经验,因此它主要是基于观点的):

  1. 我不会这样做,因为这可能会带来太多工作,导致非常脆弱的测试。如果你解决了兼容性问题,当事情出现问题时,你将很难找出真正的问题所在。
  2. 这是一种有效的方法:你可以实施一组具有可预测结果的集成测试,在Staging环境中部署你的代码,然后运行测试套件。你甚至可以在CI/CD流程中的一个步骤中运行这些测试以进行自动化。
  3. 对于本地开发来说,这是一种有效的方法:用于本地开发和测试新实现。你可以用这个进行单元测试,但每次运行后都必须进行清理(可能编写脚本以在拆卸时自动进行清理)以避免残留导致测试失败或意外通过。
  4. 这是一种有效的方法,但不太适用于测试数据库。我通常使用这个来测试API和从数据库中提取的数据的“后处理”方法。换句话说,只因为模拟工作了,并不意味着最终结果也会工作。顺便说一下,这是一篇有趣的文章
  5. 这是一种有效的方法:testcontainers提供了一个用于测试的容器化数据库,我们可以在其中添加模式和表格而无需担心(如果某些地方出现问题,意味着测试早就发现了问题),然后在测试后将其卸载。我建议在这种情况下使用这种方法!(这里有一个使用pytest的基本示例)
  6. 恐惧导致不使用测试,不使用测试导致很多不眠之夜,而很多不眠之夜会导致更多问题。这是通向黑暗的道路(还有许多失去的日子,甚至周末)。

最后,请记住,在这里没有“灵丹妙药”,所以可以结合使用不同的方法。

例如,在你的情况下,我会在本地和提交前使用testcontainers进行测试,然后在部署到生产环境之前在Staging环境中运行集成测试套件。

你对代码的方法在一瞥之下看起来不错。一个小建议是:由于你正在使用SQLModel,你可以避免使用用于“传统”FastAPI方法的单独的SQLAlchemymodels和Pydanticschemas。你可以直接使用SQLModels作为schemas,就像文档中演示的那样

希望这可以帮助你找到解决你测试问题的好方法 FastAPI 和 PostgreSQL 的测试

英文:

Testing is always a confusing matter when you begin with it, with many opinions and few things set in stone.

That said your questioning approach seems correct and thought out so I will try to answer (keep in mind that this comes from personal preference and experience on writing tests for a multitude of applications so far, so it will be mostly opinionated):

  1. I wouldn't do that because it may be too much work and result in very brittle tests. If you work around the compatibility issues, if/when things break you will have trouble finding out what the real issue is.
  2. This is a valid approach: You can implement a set of integration tests with predictable results, deploy your code in a Staging environment and run your suite against that. You can even run these tests in a step of your CI/CD pipeline for automation.
  3. That is a valid approach for local DEV: Use this for your local development and testing of new implementations. You can use this for Unit Testing, but you must cleanup after each run (probably write scripts that automate the cleanup on teardown) to avoid residues and your tests failing or passing unexpectedly.
  4. That is a valid approach but doesn't work that well for testing the DB. This I am usually using for testing the API and "postprocessing" methods on data that are fetched from the DB. In other words, just because the mocks work, doesn't mean that the end result will. Read an interesting article about this btw.
  5. This is a valid approach: testcontainers provides a containerized DB that spawns for testing, we can add schemas and tables to it without fear (if something breaks here, means that the testing caught an issue early) and then it gets despawned afterward. I would use this approach for this case! (here is a basic example of using it with pytest)
  6. Fear leads to not using tests, not using tests leads to many sleepless nights, and many sleepless nights lead to more issues. This is the path to the dark side (and many lost days and even weekends).

Finally, keep in mind that there are no "silver bullets" here so different approaches can be used together.

For example in your case, I would go with testcontainers for local and pre-commit testing and then run an integration test suite in a Staging environment before deploying to PROD.

<hr>

Your approach to the code seems good at a glance.<br/>
A minor remark here: Since you are using SQLModel you can avoid the separate SQLAlchemy models & Pydantic schemas that are used for the "traditional" FastAPI approach with SQLAlchemy.
You can use the SQLModels directly as schemas instead as demonstrated in the documentation

Hope this helps you to find a good solution for your testing FastAPI 和 PostgreSQL 的测试

huangapple
  • 本文由 发表于 2023年6月22日 17:14:07
  • 转载请务必保留本文链接:https://go.coder-hub.com/76530308.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定