如何为FastAPI项目的Alembic迁移配置pytest测试数据库环境

1. Alembic配置中url设置顺序错误，fileConfig调用会覆盖你提前设置的sqlalchemy.url参数
2. pytest fixture依赖逻辑错误，直接调用fixture函数不会触发pytest的fixture生命周期，需要作为参数声明依赖
3. 未实现Alembic的在线迁移函数，pytest运行时默认走在线迁移逻辑，缺失对应代码会导致迁移不执行
4. 没有保证全局DATABASE_URL在测试阶段指向测试库，导致迁移和业务代码仍然连接正式开发库

1. 修改项目配置层，支持测试库自动切换 #

调整app/core/config.py，添加测试环境判断逻辑，示例代码：

from pydantic_settings import BaseSettings
import os

class Settings(BaseSettings):
    DATABASE_URL: str = "postgresql://user:pass@localhost:5432/mydb"

    @property
    def SQLALCHEMY_DATABASE_URL(self):
        # 测试环境自动给库名加_test后缀
        if os.getenv("TEST") == "True":
            return f"{self.DATABASE_URL}_test"
        return self.DATABASE_URL

settings = Settings()
DATABASE_URL = settings.SQLALCHEMY_DATABASE_URL

2. 修正alembic/env.py配置 #

调整url设置顺序，补全在线迁移逻辑：

from app.core.config import DATABASE_URL
from app.db import Base
from alembic import context
from logging.config import fileConfig
from sqlalchemy import create_engine

config = context.config
# 先读配置文件，再覆盖url参数，避免被alembic.ini的默认地址覆盖
fileConfig(config.config_file_name)
config.set_main_option('sqlalchemy.url', DATABASE_URL)

target_metadata = Base.metadata

def run_migrations_offline():
    url = config.get_main_option("sqlalchemy.url")
    context.configure(
        url=url,
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    # 补全在线迁移逻辑，pytest默认走该分支
    connectable = create_engine(config.get_main_option("sqlalchemy.url"))
    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )
        with context.begin_transaction():
            context.run_migrations()

if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()

3. 修正conftest.py的fixture逻辑 #

调整fixture依赖关系，确保测试库创建后再执行迁移：

import pytest
import alembic
from alembic.config import Config
from sqlalchemy import create_engine
from sqlalchemy_utils import database_exists, create_database, drop_database
import os
import warnings

@pytest.fixture(scope="session")
def test_db_url():
    # 提前设置测试环境变量，确保后续DATABASE_URL指向测试库
    os.environ["TEST"] = "True"
    # 重新导入获取测试库地址
    from app.core.config import DATABASE_URL as TEST_DATABASE_URL
    return TEST_DATABASE_URL

@pytest.fixture(scope="session")
def create_test_db(test_db_url):
    # 清理旧测试库，创建新库
    if database_exists(test_db_url):
        drop_database(test_db_url)
    create_database(test_db_url)
    engine = create_engine(test_db_url)
    yield engine
    drop_database(test_db_url)

@pytest.fixture(scope="session")
def apply_migrations(create_test_db, test_db_url):
    # 依赖create_test_db，确保库存在后再跑迁移
    warnings.filterwarnings("ignore", category=DeprecationWarning)
    config = Config("alembic.ini")
    # 手动指定测试库地址，避免读取到ini里的正式地址
    config.set_main_option("sqlalchemy.url", test_db_url)
    alembic.command.upgrade(config, "head")
    yield
    alembic.command.downgrade(config, "base")

# 可选：添加数据库会话fixture，供测试用例直接使用
@pytest.fixture(scope="function")
def db_session(apply_migrations, create_test_db):
    from app.db import SessionLocal
    session = SessionLocal(bind=create_test_db.connect())
    yield session
    session.rollback()
    session.close()

4. 测试用例引用fixture #

在测试文件中直接引用对应fixture即可自动触发测试环境初始化：

def test_create_user(db_session):
    # 测试逻辑直接使用db_session，已自动连接测试库
    user = User(username="test")
    db_session.add(user)
    db_session.commit()
    assert user.id is not None

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