TopolearnEncryption, JSON shapes, timezone rules, auditing and multi-tenancy are usually implemented with custom types and events.
A TypeDecorator wraps an existing type and converts values on the way in and out. Set impl to the underlying type, implement process_bind_param(value, dialect) (Python to database) and process_result_value(value, dialect) (database to Python), and set cache_ok = True if the type's state is safe for the statement cache. Common uses: forcing timezone-aware UTC datetimes, storing enums or small value objects as strings, transparent encryption.
Mutation tracking is a classic gotcha: if your type stores a dict or list, in-place changes (obj.data["k"] = 1) are not detected, because the attribute wasn't reassigned. Wrap the type with MutableDict.as_mutable(...) or reassign the value.
Events hook into SQLAlchemy internals with event.listen(target, "name", fn) or @event.listens_for:
- Engine/connection:
before_cursor_execute(logging, query counting),connect(per-connection setup like SQLitePRAGMA foreign_keys=ON). - Mapper:
before_insert,before_updateper object, e.g. timestamps. - Session:
before_flushto inspectsession.new,session.dirtyandsession.deleted(auditing),after_commit. - ORM execute:
do_orm_executeto rewrite every ORM query, e.g. global tenant filters withwith_loader_criteria.
Events run synchronously inside the flush or execution, so keep them fast and avoid emitting new flushes from within flush events.
Example
import json
from sqlalchemy import create_engine, event, select, String
from sqlalchemy.types import TypeDecorator
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
class JSONText(TypeDecorator):
impl = String
cache_ok = True
def process_bind_param(self, value, dialect):
return None if value is None else json.dumps(value, sort_keys=True)
def process_result_value(self, value, dialect):
return None if value is None else json.loads(value)
class Base(DeclarativeBase): pass
class Doc(Base):
__tablename__ = "docs"
id: Mapped[int] = mapped_column(primary_key=True)
meta: Mapped[dict] = mapped_column(JSONText)
@event.listens_for(Doc, "before_insert")
def stamp(mapper, connection, target):
target.meta = {**target.meta, "v": 1}
engine = create_engine("sqlite://"); Base.metadata.create_all(engine)
with Session(engine) as s:
s.add(Doc(meta={"tag": "x"})); s.commit()
print(s.scalar(select(Doc.meta))) # {'tag': 'x', 'v': 1}Remember
- TypeDecorator: impl + process_bind_param + process_result_value
- Set cache_ok = True when the type is safe to cache
- In-place changes to dict/list columns need Mutable* wrappers
- Engine, mapper, session and ORM-execute events hook into every layer
Watch out for
- Mutating a JSON dict in place and finding the change never got saved.