使用pytest-postgresql时第二个测试仍复用首个测试的Schema问题

尝试使用pytest-postgresql进行测试，但未完全理解fixture生成的engine工作机制。以下是复现代码：两个测试用例分别对应tokyo和london，各自创建对应城市名称的Schema，并在该Schema下创建citymeta表。

import pytest
from pytest_postgresql.janitor import DatabaseJanitor
from sqlalchemy import create_engine, inspect
from sqlalchemy.orm import Session
from sqlalchemy import text
from sqlalchemy import (Column, Integer, String)
from sqlalchemy.orm import declarative_base


Base = declarative_base()


class BaseModel(Base):
    __abstract__ = True
    id = Column(Integer, primary_key=True)


class CityMeta(BaseModel):
    __tablename__ = "citymeta"
    id = Column(Integer, primary_key=True)
    version = Column(String(10), nullable=False)


def create_table(engine, schema):
    with engine.connect() as conn:
        conn.execute(text(f"""CREATE SCHEMA IF NOT EXISTS
                                {schema}"""))
        conn.commit()

    for table in BaseModel.metadata.tables.values():
        table.schema = schema
        table.create(engine, checkfirst=True)


schema1 = "tokyo"
schema2 = "london"

@pytest.fixture(scope="session")
def engine_postgresql(postgresql_proc):
    with DatabaseJanitor(
        postgresql_proc.user,
        postgresql_proc.host,
        postgresql_proc.port,
        postgresql_proc.dbname,
        postgresql_proc.version,
        password=postgresql_proc.password,
    ):
        yield create_engine(
            f"postgresql+psycopg2://{postgresql_proc.user}:"
            f"{postgresql_proc.password}@{postgresql_proc.host}:"
            f"{postgresql_proc.port}/{postgresql_proc.dbname}"
        )


def test_tokyo(engine_postgresql):
    create_table(engine_postgresql, schema1)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema1)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='2.3')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '2.3'


def test_london(engine_postgresql):
    create_table(engine_postgresql, schema2)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema2)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='0.0')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '0.0'

原本预期两个测试正常运行，但第二个测试报错：

sqlalchemy.exc.IntegrityError: (psycopg2.errors.UniqueViolation) duplicate key value violates unique constraint "citymeta_pkey"
E       DETAIL:  Key (id)=(1) already exists.
E       
E       [SQL: INSERT INTO tokyo.citymeta (id, version) VALUES (%(id)s, %(version)s)]
E       [parameters: {'id': 1, 'version': '0.0'}]
E       (Background on this error at: https://sqlalche.me/e/20/gkpj)

第二个测试虽成功创建london Schema，但会话仍使用tokyo Schema，导致id=1重复引发主键冲突。已知pytest-postgresql会维持同一数据库，为何新Schema未被新会话使用？

核心原因 #

问题并非pytest-postgresql的engine问题，而是SQLAlchemy的模型元数据是全局共享的：

• CityMeta类的__table__.schema属性在第一个测试中被设置为tokyo，这个全局设置不会自动重置。
• 第二个测试调用create_table时，虽然临时把table.schema改成了london，但SQLAlchemy的会话生成SQL时，依然会沿用之前缓存的元数据，导致后续操作还是指向tokyo Schema。

解决方案 #

方案1：动态生成测试专属模型（推荐） #

避免全局模型被修改，每次测试根据目标Schema动态生成对应的CityMeta模型：

def create_city_meta_model(schema):
    class CityMeta(BaseModel):
        __tablename__ = "citymeta"
        __table_args__ = {"schema": schema}
        id = Column(Integer, primary_key=True)
        version = Column(String(10), nullable=False)
    return CityMeta

# 修改create_table函数，不再修改全局模型的schema
def create_table(engine, schema):
    with engine.connect() as conn:
        conn.execute(text(f"""CREATE SCHEMA IF NOT EXISTS {schema}"""))
        conn.commit()
    # 动态创建模型并生成表
    CityMeta = create_city_meta_model(schema)
    CityMeta.__table__.create(engine, checkfirst=True)
    return CityMeta

# 修改测试用例
def test_tokyo(engine_postgresql):
    CityMeta = create_table(engine_postgresql, schema1)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema1)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='2.3')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '2.3'

def test_london(engine_postgresql):
    CityMeta = create_table(engine_postgresql, schema2)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema2)

    with Session(engine_postgresql) as session:
        meta = CityMeta(id=1, version='0.0')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '0.0'

方案2：使用会话级Schema映射 #

通过SQLAlchemy会话的schema_translate_map参数，动态映射模型到目标Schema，无需修改模型类：

# 修改create_table函数，不修改全局模型schema
def create_table(engine, schema):
    with engine.connect() as conn:
        conn.execute(text(f"""CREATE SCHEMA IF NOT EXISTS {schema}"""))
        conn.commit()
    # 直接指定schema创建表
    CityMeta.__table__.create(engine, checkfirst=True, schema=schema)

# 修改测试用例的会话创建逻辑
def test_tokyo(engine_postgresql):
    create_table(engine_postgresql, schema1)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema1)

    # 将模型默认的空schema映射到tokyo
    with Session(engine_postgresql, schema_translate_map={None: schema1}) as session:
        meta = CityMeta(id=1, version='2.3')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '2.3'

def test_london(engine_postgresql):
    create_table(engine_postgresql, schema2)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema2)

    # 将模型默认的空schema映射到london
    with Session(engine_postgresql, schema_translate_map={None: schema2}) as session:
        meta = CityMeta(id=1, version='0.0')
        session.add(meta)
        session.commit()
        res = session.query(CityMeta).filter_by(id=1).first()
        assert res.version == '0.0'

方案3：测试后重置模型元数据 #

如果不想改动模型结构，可以在每个测试前后手动重置模型的schema属性：

# 修改create_table函数，临时设置schema并创建表
def create_table(engine, schema):
    with engine.connect() as conn:
        conn.execute(text(f"""CREATE SCHEMA IF NOT EXISTS {schema}"""))
        conn.commit()
    # 保存原始schema，创建表后恢复
    original_schema = CityMeta.__table__.schema
    CityMeta.__table__.schema = schema
    try:
        CityMeta.__table__.create(engine, checkfirst=True)
    finally:
        CityMeta.__table__.schema = original_schema

# 修改测试用例
def test_london(engine_postgresql):
    create_table(engine_postgresql, schema2)
    insp = inspect(engine_postgresql)
    assert insp.has_table("citymeta", schema=schema2)

    with Session(engine_postgresql) as session:
        # 临时切换到london schema进行操作
        original_schema = CityMeta.__table__.schema
        CityMeta.__table__.schema = schema2
        try:
            meta = CityMeta(id=1, version='0.0')
            session.add(meta)
            session.commit()
            res = session.query(CityMeta).filter_by(id=1).first()
            assert res.version == '0.0'
        finally:
            # 恢复原始schema
            CityMeta.__table__.schema = original_schema

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