Stop loading tables you never use: lazy-alchemy v2 for SQLAlchemy 2, async, Pydantic, and SQLModel

You have a database with 120 tables. Your application only ever touches 15 of them.

SQLAlchemy doesn't care. At startup, it reflects all 120 — loading column metadata, constraints, indexes, and type information for every single one, including the 105 tables your code will never query. In large schemas this costs real time: minutes of startup delay on every deploy, every cold start, every test run.

lazy-alchemy fixes this by loading table metadata on demand — only when your code first accesses a table, and never again after that (unless you want it to).

v2.0.0 is a full modernisation: SQLAlchemy 2.x support, native asyncio, Pydantic v2 schema generation, SQLModel integration, thread-safe caching with TTL, and proper type stubs. This post walks through what changed and why it matters.

The core idea hasn't changed

The API is the same one-liner it always was:

from lazy_alchemy import get_lazy_class

from sqlalchemy import create_engine



engine  = create_engine("postgresql://user:pass@localhost/mydb")

lazy_db = get_lazy_class(engine)



# Nothing has been reflected yet.

# This line triggers reflection for 'users' — and only 'users'.

users = lazy_db.users

users is a CustomTable — a sqlalchemy.Table subclass. You can query it immediately with any SQLAlchemy expression:

from sqlalchemy import select



with engine.connect() as conn:

    rows = conn.execute(

        select(users).where(users.c.active == True)

    ).all()

The second time you access lazy_db.users, the reflection doesn't happen again. The table lives in a module-level cache keyed by (engine_url, schema, table_name).

What broke in v2 — and what we fixed

If you tried running the original lazy-alchemy against SQLAlchemy 2.x, you got hard crashes, not deprecation warnings.

MetaData(engine) was removed. The old code did self.metadata = MetaData(engine) to bind an engine globally. SA2 removed this entirely. Fix: MetaData() with no bind argument, engines passed at reflection time.

autoload=True was removed. The descriptor called Table(name, metadata, autoload=True). SA2 requires autoload_with=conn with an explicit connection. Fix:

with instance.engine.connect() as conn:

    table = CustomTable(name, metadata, autoload_with=conn)

Both fixes are backward-compatible with SA 1.4 via a version detection shim, so upgrading lazy-alchemy and SQLAlchemy at the same time works cleanly.

Async support

FastAPI, Litestar, and most modern Python services run on asyncio. Blocking an event loop for table reflection — even for 50ms — creates latency spikes across every concurrent request.

v2 adds an AsyncLazyDBAccessor that does reflection inside the event loop using asyncio.Lock:

from lazy_alchemy import get_lazy_class

from sqlalchemy.ext.asyncio import create_async_engine



engine  = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")

lazy_db = get_lazy_class(engine)   # same factory, detects AsyncEngine automatically



# In any async context:

orders = await lazy_db.get("orders")

The factory auto-detects whether you passed a sync Engine or an AsyncEngine and returns the right accessor. You don't need to import or instantiate anything different.

from sqlalchemy.ext.asyncio import AsyncSession

from sqlalchemy import select



async def list_orders(session: AsyncSession):

    orders = await lazy_db.get("orders")

    result = await session.execute(select(orders))

    return result.mappings().all()

The async cache uses double-checked locking — two concurrent coroutines hitting a cold cache will not both trigger reflection. Only one reflects; the other waits and reads the result.

Pydantic v2 schema generation

This is where things get interesting for teams working with existing databases.

Call .as_pydantic() on any reflected table and get back a Pydantic BaseModel subclass, with types and nullability inferred directly from the live schema:

lazy_db    = get_lazy_class(engine)

UserSchema = lazy_db.users.as_pydantic()



user = UserSchema(username="alice", age=30)

print(user.model_dump_json())

# {"username": "alice", "age": 30}

What gets inferred:

For PATCH endpoints where every field should be optional, .as_pydantic_partial() wraps every type in Optional:

UserUpdate = lazy_db.users.as_pydantic_partial()



patch = UserUpdate(age=31)   # username not required

SQLModel integration — working with existing tables

SQLModel's normal workflow assumes you write the model first and the database table is generated from it. Most production databases have it the other way round: the schema already exists, it's owned by someone else, and you need Python classes that work against it without duplicating every column definition by hand.

.as_sqlmodel() reflects the table and dynamically constructs the equivalent SQLModel class, with primary keys, foreign key wiring, and nullability all set correctly:

from sqlmodel import Session, select



User = lazy_db.users.as_sqlmodel()



print(User.__tablename__)   # "users"

print(User.model_fields)    # {"username": FieldInfo(primary_key=True), "age": FieldInfo(...)}



with Session(engine) as session:

    user    = session.get(User, "alice")

    results = session.exec(select(User).where(User.age > 25)).all()

FastAPI endpoint in five lines

Because User is a real SQLModel class, you can drop it straight into a FastAPI endpoint as a response_model. FastAPI uses it to generate the OpenAPI schema and validate the response — derived entirely from what's in the database:

from fastapi import FastAPI, Depends

from sqlmodel import Session, select

from lazy_alchemy import get_lazy_class



engine  = create_engine("postgresql://...")

lazy_db = get_lazy_class(engine)

app     = FastAPI()



def get_session():

    with Session(engine) as s:

        yield s



User = lazy_db.users.as_sqlmodel()



@app.get("/users/{username}", response_model=User)

def get_user(username: str, session: Session = Depends(get_session)):

    return session.get(User, username)

Zero hand-written schema code. The /docs page gets correct field types and response shapes from the database itself.

Full CRUD with mixed Pydantic + SQLModel

The cleanest pattern pairs .as_sqlmodel() for queries and responses with .as_pydantic() / .as_pydantic_partial() for request body validation:

User        = lazy_db.users.as_sqlmodel()       # queries + response_model

UserCreate  = lazy_db.users.as_pydantic()       # POST body

UserUpdate  = lazy_db.users.as_pydantic_partial() # PATCH body



@app.post("/users", response_model=User)

def create_user(body: UserCreate, session: Session = Depends(get_session)):

    user = User(**body.model_dump())

    session.add(user)

    session.commit()

    session.refresh(user)

    return user



@app.patch("/users/{username}", response_model=User)

def update_user(username: str, body: UserUpdate, session: Session = Depends(get_session)):

    user = session.get(User, username)

    for field, value in body.model_dump(exclude_none=True).items():

        setattr(user, field, value)

    session.commit()

    session.refresh(user)

    return user

Production-grade caching

The v1 cache lived on the descriptor itself — one _table reference per descriptor instance. This meant multiple get_lazy_class() calls on the same engine would each reflect the same tables independently.

v2 moves the cache to module level, keyed by (engine_url, schema, table_name):

# Both of these share a single cache entry for "users"

lazy_db_1 = get_lazy_class(engine)

lazy_db_2 = get_lazy_class(engine)

Additional controls:

# Re-reflect after 5 minutes (useful for long-running services with live migrations)

lazy_db = get_lazy_class(engine, cache_ttl=300)



# Invalidate one table

lazy_db.invalidate("users")



# Wipe everything for this engine

lazy_db.invalidate_all()



# Warm the cache at startup

lazy_db.preload("users", "orders", "products")



# List all tables in the schema

lazy_db.list_tables()

# → ['users', 'orders', 'products', 'order_items', ...]

Thread safety uses double-checked locking:

# Fast path — no lock

cached = _get_cached_table(key, ttl)

if cached: return cached



# Slow path — acquire lock, re-check, then reflect

with _TABLE_CACHE_LOCK:

    cached = _get_cached_table(key, ttl)

    if cached: return cached

    table = reflect(...)

    _set_cached_table(key, table)

    return table

Multi-schema support

public_db    = get_lazy_class(engine, schema="public")

analytics_db = get_lazy_class(engine, schema="analytics")



users        = public_db.users

monthly_kpis = analytics_db.monthly_kpis

Each accessor maintains its own isolated cache namespace. Tables from different schemas with the same name don't collide.

Meaningful error messages

The old behavior: a cryptic SA internals traceback when you accessed a table that didn't exist.

The new behavior:

TableNotFoundError: Table 'usr' not found in database 'mydb'.

Available tables: order_items, orders, products, users

from lazy_alchemy import LazyAlchemyError, TableNotFoundError, ReflectionError



try:

    table = lazy_db.usr

except TableNotFoundError as e:

    print(e)  # includes the list of available tables

Type safety

The package ships py.typed and __init__.pyi stubs, so mypy and pyright understand the full API surface without any plugin or extra config:

from lazy_alchemy import get_lazy_class, LazyDB, AsyncLazyDBAccessor, CustomTable



lazy_db : LazyDB              = get_lazy_class(sync_engine)

accessor: AsyncLazyDBAccessor = get_lazy_class(async_engine)

table   : CustomTable         = lazy_db.users

Installation

pip install lazy-alchemy

Everything — async, Pydantic, SQLModel — is included in the base install. Python ≥ 3.10, SQLAlchemy ≥ 2.0.

# Development

pip install "lazy-alchemy[dev]"

pytest

Summary

If you work with large SQLAlchemy schemas, async Python services, or FastAPI over existing databases — give it a try and let me know what you think.

GitHub: github.com/satyamsoni2211/lazy_alchemy

PyPI: pypi.org/project/lazy-alchemy