# Before: every table reflected at startup — even the ones you never touch
engine = create_engine(DATABASE_URL)        # ← hundreds of tables loaded

# After: only the tables you actually use, only when you use them
lazy_db = get_lazy_class(engine)
table   = lazy_db.users                     # ← reflected here, on first access

Why lazy-alchemy?

SQLAlchemy reflects every table in your database when metadata is loaded at startup. In large schemas — think 100+ tables across microservices or multi-tenant systems — this can stretch application startup from seconds into minutes, and waste memory on models you may never query.

lazy-alchemy defers reflection entirely. Tables are loaded from the database the first time you access them, then cached. Your app starts instantly. The rest just works.

v2.0.0 is a complete modernisation: SQLAlchemy 2.x support, native asyncio, Pydantic v2 schema generation, SQLModel integration, thread-safe caching with TTL, and full type stubs.

Architecture

The diagram below shows how all components relate — from the unified get_lazy_class entry point, through the sync and async engine paths, down through the shared module-level cache and SA2 reflection layer, to the CustomTable output methods.

The lifecycle diagram shows exactly what happens at runtime on every table access — the fast cache hit path, the full reflection path (with TableNotFoundError guard) when the cache misses, and the parallel async path.

Installation

pip install lazy-alchemy

Requires: Python ≥ 3.10 · SQLAlchemy ≥ 2.0 · Pydantic ≥ 2.0 · SQLModel ≥ 0.0.16

All features — async support, Pydantic schema generation, and SQLModel integration — are included in the base install.

For development and testing:

pip install "lazy-alchemy[dev]"

Quick start

Sync

from lazy_alchemy import get_lazy_class
from sqlalchemy import create_engine, select

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

# Table is reflected once, on first access, then cached
users = lazy_db.users

with engine.connect() as conn:
    rows = conn.execute(select(users).where(users.c.active == True)).all()

Async

from lazy_alchemy import get_lazy_class
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

engine  = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
lazy_db = get_lazy_class(engine)       # detects AsyncEngine automatically

async def get_users(session: AsyncSession):
    users  = await lazy_db.get("users")
    result = await session.execute(select(users))
    return result.mappings().all()

The same get_lazy_class factory works for both sync and async engines — it auto-detects the engine type and returns the right accessor.

Features

SQLAlchemy 2.x compatible

lazy-alchemy v2 uses the SA2-native reflection API throughout:

# Uses autoload_with=conn (SA2) instead of the removed autoload=True (SA1)
# Uses MetaData() without bind — engines are passed at reflection time
lazy_db = get_lazy_class(engine)
table   = lazy_db.orders              # reflected via autoload_with on first access

Async-native

Pass an AsyncEngine and get a non-blocking accessor. Reflection happens inside the event loop with asyncio.Lock protection — no thread executor needed.

engine  = create_async_engine("postgresql+asyncpg://...")
lazy_db = get_lazy_class(engine)

# In any async context:
orders = await lazy_db.get("orders")

# Async tables also support schema generation
Order = orders.as_sqlmodel()

Pydantic v2 schema generation

Reflect a table and get a validated Pydantic model in one line — column types, nullability, and defaults are all inferred from the live schema.

lazy_db    = get_lazy_class(engine)
UserSchema = lazy_db.users.as_pydantic()

# Full model with type validation
user = UserSchema(id=1, name="Alice", email="alice@example.com")
print(user.model_dump_json())

# Partial schema — all fields Optional, useful for PATCH endpoints
UserPatch = lazy_db.users.as_pydantic_partial()
patch     = UserPatch(name="Alice Updated")   # everything else is None

SQLModel integration

Generate a first-class SQLModel class — usable with FastAPI response models, OpenAPI schema generation, and SQLModel sessions.

from sqlmodel import Session, select

User    = lazy_db.users.as_sqlmodel()
session = Session(engine)

# Works as a FastAPI response_model, query target, and Pydantic model
@app.get("/users/{user_id}", response_model=User)
def get_user(user_id: int):
    return session.get(User, user_id)

Thread-safe caching with TTL

Reflected tables are stored in a module-level cache shared across all instances for the same engine. A double-checked lock prevents duplicate reflection under concurrent load.

# Re-reflect tables after 5 minutes (useful after live migrations)
lazy_db = get_lazy_class(engine, cache_ttl=300)

# Force a specific table to re-reflect on next access
lazy_db.invalidate("users")

# Wipe the entire cache for this engine
lazy_db.invalidate_all()

# Warm the cache at startup for your hottest tables
lazy_db.preload("users", "orders", "products")

Multi-schema support

# PostgreSQL schema namespacing
public_db    = get_lazy_class(engine, schema="public")
analytics_db = get_lazy_class(engine, schema="analytics")

users  = public_db.users
events = analytics_db.events

Table introspection

# List every table in the schema
lazy_db.list_tables()
# → ['users', 'orders', 'products', 'order_items', ...]

# Column access via attribute delegation
table = lazy_db.users
print(table.email)     # equivalent to table.c.email

Meaningful errors

lazy_db.nonexistent_table
# TableNotFoundError: Table 'nonexistent_table' not found in database 'mydb'.
# Available tables: order_items, orders, products, users

Full type safety

The package ships a py.typed marker and typed stubs (__init__.pyi), so mypy and pyright understand the full API surface with no extra configuration.

# mypy / pyright understand all of this:
lazy_db : LazyDB              = get_lazy_class(sync_engine)
accessor: AsyncLazyDBAccessor = get_lazy_class(async_engine)
table   : CustomTable         = lazy_db.users
schema  : type[BaseModel]     = table.as_pydantic()

API reference

get_lazy_class(engine, *, cache_ttl=None, schema=None)

Unified factory. Returns a LazyDB for sync engines and an AsyncLazyDBAccessor for async engines.

LazyDB (sync)

AsyncLazyDBAccessor (async)

CustomTable

A sqlalchemy.Table subclass with extra convenience methods.

Exceptions

from lazy_alchemy import LazyAlchemyError, TableNotFoundError, ReflectionError

Recipes

FastAPI with async sessions

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import select
from lazy_alchemy import get_lazy_class

engine       = create_async_engine("postgresql+asyncpg://user:pass@localhost/mydb")
lazy_db      = get_lazy_class(engine)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)

@asynccontextmanager
async def lifespan(app: FastAPI):
    # Warm the cache for your hottest tables at startup
    await lazy_db.get("users")
    await lazy_db.get("orders")
    yield

app = FastAPI(lifespan=lifespan)

async def get_session():
    async with SessionLocal() as session:
        yield session

@app.get("/users/{user_id}")
async def get_user(user_id: int, session: AsyncSession = Depends(get_session)):
    users  = await lazy_db.get("users")
    result = await session.execute(select(users).where(users.c.id == user_id))
    return result.mappings().first()

CRUD endpoints with Pydantic schemas

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)

# Full schema for POST / response model
UserCreate = lazy_db.users.as_pydantic()

# Partial schema for PATCH — only provided fields are updated
UserUpdate = lazy_db.users.as_pydantic_partial()

@app.post("/users", response_model=UserCreate)
def create_user(body: UserCreate): ...

@app.patch("/users/{user_id}", response_model=UserCreate)
def update_user(user_id: int, body: UserUpdate): ...

Cache invalidation after migrations

from alembic import context

def run_migrations_online():
    with engine.connect() as connection:
        context.configure(connection=connection)
        with context.begin_transaction():
            context.run_migrations()

    # Invalidate after migration so next access re-reflects new schema
    lazy_db.invalidate_all()

Working with multiple schemas

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

public    = get_lazy_class(engine, schema="public")
reporting = get_lazy_class(engine, schema="reporting")
archive   = get_lazy_class(engine, schema="archive")

# Each accessor has its own isolated cache
users         = public.users
monthly_sales = reporting.monthly_sales
legacy_orders = archive.orders_2019

Migrating from v1

v2 requires SQLAlchemy 2.x. If you are still on SQLAlchemy 1.4, stay on lazy-alchemy v1 until you are ready to upgrade both at the same time.

# v1 — still works, no changes needed on your side
from lazy_alchemy import get_lazy_class
from sqlalchemy import create_engine

lazy_db  = get_lazy_class(create_engine(DB_URL))
db_model = lazy_db.my_table
rows     = session.query(db_model).filter(db_model.foo == "bar").all()

When you upgrade to SQLAlchemy 2.x, the new-style query API is recommended:

# SQLAlchemy 2.x style
from sqlalchemy import select

rows = session.execute(
    select(db_model).where(db_model.c.foo == "bar")
).all()

What changed in v2

Development

git clone https://github.com/satyamsoni2211/lazy_alchemy
cd lazy_alchemy
pip install -e ".[dev]"

# Run tests
pytest

# Run with coverage
pytest --cov=lazy_alchemy --cov-report=term-missing

# Type checking
mypy lazy_alchemy

# Linting
ruff check lazy_alchemy

License

Released under the MIT License.