Sqlalchemy 2.0异步环境下存在表却报'relation x does not exist'

我已经把问题简化为最基础的测试用例，这是测试模拟流程的一部分：db_session已映射到测试专属schema，我确认环境搭建正常且该schema包含所需的organization表，但插入记录时出现问题。

我完全搞不懂：既然查询能找到表，为什么插入会提示表不存在？而且用的是同一个数据库连接，不存在权限或范围差异。

测试代码 #

async def generate(self) -> "Any":
    tables = await self.db_session.execute(text("SELECT table_name FROM test_hero.information_schema.tables where table_schema = 'test_smoke_respond_to_message'"))
    if ("organization",) in tables and self.db_session.get_bind().get_execution_options()["schema_translate_map"]["public"] == "test_smoke_respond_to_message":
        org = SqlOrganization(**self.model_dict)
        async with self.db_session as session:
            session.add(org)
            await session.commit()
            await session.refresh(org)

执行错误信息 #

执行await session.commit()时抛出以下错误：

sqlalchemy.exc.ProgrammingError: (sqlalchemy.dialects.postgresql.asyncpg.ProgrammingError) <class 'asyncpg.exceptions.UndefinedTableError'>: relation "organization" does not exist
[SQL: INSERT INTO organization (h_engine_id, natural_language_name, _id) VALUES ($1::VARCHAR, $2::VARCHAR, $3::UUID) RETURNING organization.created_at, organization.updated_at]
[parameters: ('0fda47a9dfe14d02a6a70e2fd15ddfec', 'Archer Inc', UUID('7b0d6120-6255-4c21-8276-4cd289c7d339'))]
> /app/app/tests/mock_factory/models.py(51)generate()

相关代码定义（已移除无关方法） #

models.py #

class SqlOrganization(NamedBase, TimestampedBase):
    """the highest level of hierarchy in our system."""

    __tablename__ = "organization"
    __pydantic_model__ = "PyOrganization"

    id_prefix = "o"
    engine_id: Mapped[Optional[str]] = mapped_column(
        String(255), nullable=True, doc="the id of the organization in the Engine"
    )

    platforms: Mapped[List["SqlPlatform"]] = relationship(
        "SqlPlatform", back_populates="organization", lazy="selectin"
    )

bases.py #

class SqlalchemyBase(AsyncAttrs, DeclarativeBase):
    __abstract__ = True
    @property
    def __pydantic_model__(self) -> "str":
        """str representation of the pydantic model"""
        raise NotImplementedError

class NamedBase(SqlalchemyBase):
    __abstract__ = True
    natural_language_name: Mapped[str] = mapped_column(String(255))

conftest.py #

@pytest_asyncio.fixture(scope="function")
async def test_async_sessionmaker(request):
    function_ = request.node.name
    engine = get_async_engine()
    if "[" in function_:
        function_ = function_.split("[")[0]
    async with engine.begin() as connection:
        await connection.execute(text('CREATE EXTENSION IF NOT EXISTS "vector"'))
        await connection.execute(text(f"CREATE SCHEMA IF NOT EXISTS {function_}"))
        await connection.execute(text(f"SET search_path TO {function_},public"))
        await connection.run_sync(SqlalchemyBase.metadata.drop_all)
        await connection.run_sync(SqlalchemyBase.metadata.create_all)

        await connection.execution_options(schema_translate_map={"public": function_})
    scoped_engine = get_async_engine(execution_options={"schema_translate_map": {"public": function_}})
    def _test_async_sessionmaker() -> "sessionmaker[AsyncSession]":
        return sessionmaker(bind=scoped_engine, class_=AsyncSession, expire_on_commit=False)
    return _test_async_sessionmaker

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