FastAPI+PostgreSQL+SQLModel多Schema项目测试最佳实践咨询

我正在开发基于FastAPI并连接PostgreSQL数据库的后端应用，使用SQLModel作为ORM工具。SQLModel文档推荐用内存SQLite做测试，但我的生产数据库使用了非public的Schema（比如示例中的pouetpouet），测试时出现了“schema pouetpouet不存在”的兼容性问题。

我需要同时开展单元测试与集成测试，目前纠结以下几种测试方案，希望得到最优建议：

• 解决生产PostgreSQL与内存SQLite的兼容性问题？
• 在预生产PostgreSQL执行测试并清理新增/删除数据？（目前在用，但认为不是最佳实践）
• 在Docker容器中搭建本地PostgreSQL服务器？
• 在pytest测试文件中用字典模拟数据库？
• 使用testcontainers等第三方库？
• 不进行测试？

项目简化版本 #

项目架构（每个文件夹均含__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")  # 参考SQLModel的Issue：#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

代码改进建议 #

1. 环境配置隔离：

   • 测试环境应使用独立的数据库配置，不要与生产/开发环境共用。可以在core/config.py中根据环境变量（比如ENVIRONMENT=test）加载不同的数据库URL。

2. Schema定义优化：

   • SQLModel现在已经支持通过__table_args__直接指定Schema，建议替换掉当前的metadata方式，代码更直观：

     class Item(SQLModel, table=True):
         __tablename__ = "cities"
         __table_args__ = {"schema": "pouetpouet"}
     
         id: int = Field(primary_key=True, default=None)
         city_name: str
     

3. 测试Session替换：

   • 当前测试直接使用生产环境的Session，会污染数据库。需要用pytest fixture替换get_session依赖，使用测试数据库的Session。

4. Schema的ORM模式简化：

   • SQLModel的ItemRead不需要手动设置orm_mode=True，SQLModel本身已经内置了ORM模式支持，可以删除该配置。

测试方案建议 #

单元测试：优先用模拟数据库（方案4） #

单元测试的核心是验证业务逻辑，不需要连接真实数据库。可以用unittest.mock替换Session，或者用字典模拟数据存储：

from unittest.mock import Mock, MagicMock
from api.crud.items import get_item, create_new_item
from api.schemas import ItemCreate

def test_get_item():
    mock_session = Mock()
    mock_item = MagicMock()
    mock_item.id = 1
    mock_item.city_name = "Paris"
    mock_session.exec.return_value.first.return_value = mock_item

    result = get_item(mock_session, 1)
    assert result.id == 1
    assert result.city_name == "Paris"
    mock_session.exec.assert_called_once()

def test_create_new_item():
    mock_session = Mock()
    item_create = ItemCreate(city_name="London")
    result = create_new_item(mock_session, obj_input=item_create)
    
    mock_session.add.assert_called_once()
    mock_session.commit.assert_called_once()
    mock_session.refresh.assert_called_once()

集成测试：优先用testcontainers（方案5） #

testcontainers可以自动启动PostgreSQL容器，测试完成后自动销毁，完全模拟生产环境，且不会污染现有数据库。实现步骤：

1. 安装依赖：pip install testcontainers-postgresql sqlalchemy
2. 编写pytest fixture：

   import pytest
   from sqlmodel import create_engine, Session, SQLModel
   from fastapi.testclient import TestClient
   from api.main import app
   from api.db.session import get_session
   from testcontainers.postgres import PostgresContainer
   
   @pytest.fixture(scope="module")
   def test_db():
       with PostgresContainer("postgres:15") as container:
           engine = create_engine(container.get_connection_url())
           # 创建Schema
           with engine.connect() as conn:
               conn.execute("CREATE SCHEMA pouetpouet;")
               conn.commit()
           # 创建表
           SQLModel.metadata.create_all(engine)
           yield engine
   
   @pytest.fixture(scope="module")
   def test_session(test_db):
       with Session(test_db) as session:
           yield session
   
   @pytest.fixture(scope="module")
   def client(test_session):
       def override_get_session():
           yield test_session
       app.dependency_overrides[get_session] = override_get_session
       yield TestClient(app)
       app.dependency_overrides.clear()
   

3. 编写集成测试：

   def test_create_item(client):
       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
   
   def test_read_item(client):
       # 先创建一个item
       create_response = client.post("/api/items/", json={"city_name": "New York"})
       item_id = create_response.json()["id"]
       # 读取该item
       response = client.get(f"/api/items/{item_id}")
       assert response.status_code == 200
       assert response.json()["city_name"] == "New York"
   

其他方案评估 #

• 方案1（SQLite兼容性）：不推荐，SQLite对PostgreSQL的Schema支持有限，且后续遇到PostgreSQL特有的语法（比如数组、JSONB）时会出现更多兼容性问题，维护成本极高。
• 方案2（预生产库测试）：不推荐，即使清理数据，也存在误删生产数据的风险，且测试会互相干扰，无法保证测试隔离。
• 方案3（Docker本地搭建）：可以用，但需要手动管理容器的启动、初始化、销毁，不如testcontainers自动化程度高，适合没有CI/CD的本地开发场景。
• 方案6（不测试）：绝对不可取，后端应用没有测试会导致上线后频繁出问题，维护成本陡增。

内容的提问来源于stack exchange，提问作者FloCAD