[FIXED] "relation x does not exist" in Sqlalchemy 2.0 Async when it very clearly does

Issue

I have tried to boil this down to the most basic of test cases. This is part of our test mocking; the db_session is mapped to a test-specific schema, I confirm that the setup works and the test-specific schema has the the required table, and then add a record to that table.

    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)

The problem is that await session.commit() throws this error:

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'))]
(Background on this error at: https://sqlalche.me/e/20/f405)
> /app/app/tests/mock_factory/models.py(51)generate()

I am completely out of ideas. How can this possibly execute if the table does not exist? How can it possibly fail if the table does exist? This is using the same DB connection so there's no possibility of different permissions/scopes.

Here are the related definitions (most of the extraneous methods removed for clarity)

# 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

Solution

So...RTFM. The secret is in schema_translate_map, which functions very differently than I understood it to function. I had this:

scoped_engine = get_async_engine(
    execution_options={"schema_translate_map": {"public": function_}}
)

Which if you look at this example from the docs clearly states that when setting to {"public": None}

Table objects with schema="public" will render with no schema

this is the opposite of what I believed it was doing - I wanted to map objects with no schema (so "public" schema by default) to a schema matching the pytest function name. What I needed to do that was this:

scoped_engine = get_async_engine(
    execution_options={"schema_translate_map": {None: function_}}
)

which is very clearly documented in the Sqlalchemy 2.0 docs. It also reads correctly in English: "when the schema is None, use this schema."

Answered By - EthanK