如何在不修改业务代码的前提下为pytest配置内存数据库以避免测试数据写入真实数据库

嘿，这个场景我太熟了——用SQLAlchemy异步写业务，想跑pytest又怕污染真实数据库，还不想动已经写好的业务代码，对吧？刚好有几个完美适配的方案，全都是不用改业务逻辑就能搞定的：

方案一：用pytest Fixture替换数据库配置（最优雅推荐） #

核心思路就是用pytest的fixture偷偷把业务代码里的数据库连接地址换成内存数据库的，让测试全程在内存里跑，真实数据库完全碰不到。

步骤1：准备依赖 #

先装异步SQLite的驱动，因为内存数据库我们用SQLite的内存模式：

pip install aiosqlite

步骤2：写测试用的Fixture #

假设你的业务代码里，数据库URL存在your_app/config.py的DATABASE_URL里，SQLAlchemy的模型基类是your_app/models.py里的Base，那可以写这几个fixture：

import pytest
import asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from your_app.config import DATABASE_URL
from your_app.models import Base
from your_app.db import DatabaseConnection  # 你的数据库连接类

# 给异步测试提供事件循环
@pytest.fixture(scope="session")
def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()

# 创建内存数据库引擎，会话级别（整个测试会话只初始化一次表）
@pytest.fixture(scope="session")
async def async_test_engine():
    test_engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
    # 初始化所有表结构
    async with test_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    yield test_engine
    # 测试结束后销毁引擎
    await test_engine.dispose()

# 每个测试用例一个独立的数据库会话，自动回滚
@pytest.fixture(scope="function")
async def db_session(async_test_engine):
    async_session = sessionmaker(
        async_test_engine, class_=AsyncSession, expire_on_commit=False
    )
    async with async_session() as session:
        async with session.begin():
            yield session
        # 测试结束自动回滚，避免脏数据影响下一个用例
        await session.rollback()

# 自动替换业务代码里的数据库URL为内存库地址
@pytest.fixture(scope="session", autouse=True)
def patch_database_url(monkeypatch):
    monkeypatch.setattr("your_app.config.DATABASE_URL", "sqlite+aiosqlite:///:memory:")

这样一来，你的DatabaseConnection类在测试时会自动读取被替换后的内存库URL，完全不用改业务里的__aenter__和__aexit__逻辑——就像Django TestCase那样，跑测试时自动切换到测试库，跑完就销毁。

方案二：直接Patch数据库连接类（适配硬编码URL的情况） #

如果你的业务代码里没把数据库URL抽成配置，而是直接硬编码在DatabaseConnection里了，那也不用改业务代码，用pytest的monkeypatch直接替换它的连接逻辑就行：

@pytest.fixture(scope="function", autouse=True)
async def patch_db_connection(monkeypatch, async_test_engine):
    # 保存原来的__aenter__方法
    original_aenter = DatabaseConnection.__aenter__

    # 写一个patched的__aenter__，用内存数据库的会话
    async def patched_aenter(self):
        self.engine = async_test_engine
        self.session = AsyncSession(async_test_engine)
        return self.session

    # 替换掉原来的方法
    monkeypatch.setattr(DatabaseConnection, "__aenter__", patched_aenter)

这个方法虽然不如配置替换优雅，但胜在完全不用碰业务代码，适合快速适配老代码。

几个注意点 #

• 内存SQLite必须用aiosqlite驱动，普通的sqlite驱动不支持异步操作
• 如果你的业务用了PostgreSQL等特定数据库的语法（比如JSONB、数组操作），SQLite可能不兼容，这时候可以用testcontainers启动一个临时的真实数据库容器，但这个需要装额外依赖，不过也不用改业务代码
• 每个测试用例用独立的会话或者自动回滚，能避免测试之间的脏数据污染，保证测试的独立性

这样配置完，你的测试就会完全在内存数据库里跑，真实数据库一点数据都不会写，和Django TestCase的体验一模一样！

内容来源于stack exchange