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:
| Column trait | Pydantic field |
|---|---|
NOT NULL |
Required field (...) |
NULL |
Optional[T] with default=None
|
INTEGER |
int |
VARCHAR / TEXT
|
str |
BOOLEAN |
bool |
TIMESTAMP / DATETIME
|
datetime.datetime |
UUID |
uuid.UUID |
JSON |
Any |
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
| v1 | v2 | |
|---|---|---|
| SQLAlchemy | 1.x only | 2.x (required) |
| Async | ✗ | ✓ AsyncEngine + asyncio.Lock |
| Pydantic | ✗ | ✓ as_pydantic(), as_pydantic_partial() |
| SQLModel | ✗ | ✓ as_sqlmodel() |
| Cache | Per-descriptor | Module-level, thread-safe, TTL |
| Type stubs | ✗ | ✓ py.typed + init.pyi |
| Multi-schema | ✗ | ✓ schema= parameter |
| Error messages | SA internals | TableNotFoundError + available tables |
| Python | ≥ 3.6 | ≥ 3.10 |
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
Top comments (0)