Mapping classes to existing tables with reflection

Reflect through await conn.run_sync(...) because reflection is synchronous, use automap or Core Table objects for exploration and scripts, and for an application declare explicit Mapped[] models and add a test that compares them with the real schema — so start-up needs no database and drift fails a build. This guide belongs to Core vs ORM architecture decisions.

Quick Answer

Reflection is synchronous, so under an async engine it has to run inside run_sync().

Before — reflecting on an async connection directly:

from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://shop:secret@legacy/shop")
metadata = MetaData()

async with engine.connect() as conn:
    metadata.reflect(bind=conn)
# sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call
# await_only() here.

After — reflection inside run_sync, for the tables you need:

from sqlalchemy import MetaData, select
from sqlalchemy.ext.asyncio import create_async_engine

engine = create_async_engine("postgresql+asyncpg://shop:secret@legacy/shop")
metadata = MetaData()

TABLES = ["customers", "orders", "order_lines"]


async def reflect_schema() -> MetaData:
    async with engine.connect() as conn:
        await conn.run_sync(lambda sync_conn: metadata.reflect(bind=sync_conn, only=TABLES))
    return metadata


metadata = await reflect_schema()
customers = metadata.tables["customers"]

rows = (await session.execute(
    select(customers.c.id, customers.c.email).where(customers.c.active.is_(True))
)).mappings().all()

only= matters on a large legacy database: reflection issues catalogue queries per table, so reflecting four hundred tables to use three is slow enough to notice at every start-up.

Execution Context & Async Workflow Integration

Reflection reads the database catalogue and builds Table objects describing what it found: columns with types, primary keys, foreign keys, indexes and constraints. All of it is implemented with SQLAlchemy's synchronous inspector, which is why an async engine has to reach it through run_sync() — the same bridge Alembic uses, described in setting up Alembic env.py for asyncpg.

Four approaches sit on top of that, and the right one depends on how long the code will live.

Core reflection — MetaData.reflect() or Table("orders", metadata, autoload_with=conn) — gives tables and no classes. For a migration script, a data-cleaning job or an ad-hoc query against a legacy database, it is exactly enough.

Automap generates mapped classes and relationships by inspecting foreign keys:

from sqlalchemy.ext.automap import automap_base

Base = automap_base()


async def prepare() -> None:
    async with engine.connect() as conn:
        await conn.run_sync(lambda sync_conn: Base.prepare(autoload_with=sync_conn))


Customer = Base.classes.customers      # named after the table

It is the fastest way to start querying an unfamiliar database, and it has two properties that make it awkward to keep: names come from the schema rather than from the domain, and a class only appears if its table has a primary key.

DeferredReflection lets classes be declared with behaviour but no columns, filled in from the database at start-up. It keeps the class names and methods under your control while still requiring the database at boot.

Explicit models declare every column, which is more typing and the only option that gives a type checker something to work with, lets the application start without the database, and makes the schema visible in code review. For anything long-lived, this is the choice — with a test that verifies the declaration against the real schema, so the duplication cannot drift.

The async caveat that catches people is start-up ordering. Reflection or automap at import time cannot work, because there is no running event loop and no engine yet; it has to happen in the lifespan, which means the application cannot serve traffic until the database has answered several catalogue queries. That is a real availability coupling: a database slow to respond at deploy time becomes an application that fails to start.

Resolving Warnings, Errors & Common Mistakes

Mapping a view is the most common reason to reach for reflection deliberately, and it needs one extra declaration. The view has no primary key, so the mapper must be told which columns identify a row:

from sqlalchemy import Table
from sqlalchemy.orm import DeclarativeBase


class Base(DeclarativeBase):
    pass


def define_order_summary(sync_conn) -> type:
    summary = Table("v_order_summary", Base.metadata, autoload_with=sync_conn)

    class OrderSummary(Base):
        __table__ = summary
        __mapper_args__ = {"primary_key": [summary.c.order_id]}

    return OrderSummary


async def load_view() -> type:
    async with engine.connect() as conn:
        return await conn.run_sync(define_order_summary)

The declared primary key does not have to be unique in the database — SQLAlchemy only needs it to identify rows in the identity map — but if it is not actually unique, two different rows collapse into one object. Choosing a genuinely unique column, or a composite, is what keeps that from happening.

A view mapped this way is read-only in practice: an update would need the view to be updatable and the mapping to know which table to write. Declaring viewonly-style intent explicitly — no setters, and a comment — avoids the surprise, and the related problem of a filtered or derived collection being treated as writable is covered in modeling relationships, cascades and association objects.

Advanced: Verifying Declared Models Against the Real Schema

Declaring models explicitly over a schema someone else owns creates a duplication: the truth is in the database, and the code claims to describe it. A test removes the risk by comparing the two, and it is short enough to write once for every model.

import pytest
from sqlalchemy import inspect

from legacy.models import Base

# Types SQLAlchemy reflects differently from how we declare them, with a reason.
ACCEPTED = {("customers", "extra"): {"JSON", "JSONB"}}


def _schema_snapshot(sync_conn) -> dict[str, dict[str, str]]:
    inspector = inspect(sync_conn)
    return {
        table: {col["name"]: type(col["type"]).__name__
                for col in inspector.get_columns(table)}
        for table in inspector.get_table_names()
    }


@pytest.mark.asyncio
async def test_models_match_the_database(engine):
    async with engine.connect() as conn:
        actual = await conn.run_sync(_schema_snapshot)

    for table in Base.metadata.sorted_tables:
        assert table.name in actual, f"{table.name} is missing from the database"
        columns = actual[table.name]
        for column in table.columns:
            assert column.name in columns, (
                f"{table.name}.{column.name} is declared but not in the database"
            )
            declared = type(column.type).__name__
            reflected = columns[column.name]
            allowed = ACCEPTED.get((table.name, column.name), {declared})
            assert reflected in allowed, (
                f"{table.name}.{column.name}: declared {declared}, database has {reflected}"
            )

Three properties make it worth having. It fails when the other team adds a NOT NULL column your models do not know about — before that column breaks an insert in production. It documents the accepted mismatches in one dictionary, with a reason, rather than as folklore. And it costs one catalogue read per table, once, in CI.

alembic check is the equivalent when you own the schema, and the two are complementary: alembic check asserts that the database matches the models by proposing migrations, while this test asserts that the models match a database you cannot migrate.

For a database you genuinely do not control, one more safeguard helps: reflect it in CI and commit the snapshot. A diff in that file during a routine build is how you learn that the upstream team changed a column type, rather than learning it from an exception:

import json
import pathlib

snapshot = await conn.run_sync(_schema_snapshot)
pathlib.Path("tests/legacy_schema.json").write_text(json.dumps(snapshot, indent=2, sort_keys=True))

Generating that file in a scheduled job and failing when it differs from the committed copy turns an external schema change into a notification with a diff — which is as close to a contract as a shared database allows.

Working With a Database Someone Else Owns

Reflection is usually a symptom: the schema belongs to another team, another application, or a vendor product, and cannot be changed to suit you. That constraint shapes more than the mapping.

Read what you need, not the whole table. A legacy table with eighty columns does not have to become a model with eighty attributes. Map the ten you use — SQLAlchemy is content with a partial mapping as long as the primary key is present and the unmapped columns are nullable or have defaults — or select columns rather than entities and skip the mapping entirely.

Never write what you do not own. A shared table often has triggers, downstream consumers and validation living in another application. If writes are yours to make, make them narrow: set the specific columns, in a set-based update(), rather than flushing an object whose other attributes might overwrite something. The ORM writes every changed column, and a model with a stale default can silently reset a field another system maintains.

Treat the schema as an external interface. That means a version-controlled snapshot, a test that checks it, and a place in the codebase where the assumptions are written down: which columns are read, which are written, what the primary key is, and which mismatches are accepted.

Prefer views for a stable contract. If the other team will create one, a view is the cleanest boundary available in a shared database: they can restructure underlying tables while keeping the view's shape, and your mapping depends only on the view. Mapping one needs the explicit primary key shown above.

Keep reflection out of the request path. automap and DeferredReflection need the database at start-up, which couples your availability to theirs. Explicit models do not, and for a service that must start during a database incident — to serve cached responses, or simply to report itself unhealthy properly — that difference matters.

Where reflection genuinely belongs is in the tools around the application: a one-off migration script that copies data out of a legacy schema, a report generator pointed at a warehouse, an exploratory notebook. In those, automap in three lines is a real saving, and none of the long-term costs apply — as long as the code stays a script and does not quietly become part of the service. The general question of which layer to use, and how to mix them, is the subject of the parent guide on Core vs ORM architecture decisions.

Frequently Asked Questions

Related

• Core vs ORM Architecture Decisions — The parent guide: choosing a layer, and mixing them.
• Executing raw SQL safely with text() and bindparams — Querying a foreign schema without mapping it at all.
• Excluding tables and schemas from Alembic autogenerate — Keeping tables you do not own out of your migrations.
• Using mapped_column() instead of Column() in SQLAlchemy 2.0 — Declaring the models this guide recommends.