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