Models¶
Added in v2.0
Reusable SQLAlchemy 2.0 mixins for common column patterns, designed to be composed freely on any DeclarativeBase model.
Overview¶
The models module provides mixins that each add a single, well-defined column behaviour. They work with standard SQLAlchemy 2.0 declarative syntax and are fully compatible with AsyncSession.
from fastapi_toolsets.models import UUIDMixin, TimestampMixin
class Article(Base, UUIDMixin, TimestampMixin):
__tablename__ = "articles"
title: Mapped[str]
content: Mapped[str]
All timestamp columns are timezone-aware (TIMESTAMPTZ). All defaults are server-side (clock_timestamp()), so they are also applied when inserting rows via raw SQL outside the ORM.
Mixins¶
UUIDMixin¶
Adds a id: UUID primary key generated server-side by PostgreSQL using gen_random_uuid(). The value is retrieved via RETURNING after insert, so it is available on the Python object immediately after flush().
Requires PostgreSQL 13+
from fastapi_toolsets.models import UUIDMixin
class User(Base, UUIDMixin):
__tablename__ = "users"
username: Mapped[str]
# id is None before flush
user = User(username="alice")
session.add(user)
await session.flush()
print(user.id) # UUID('...')
UUIDv7Mixin¶
Added in v2.3
Adds a id: UUID primary key generated server-side by PostgreSQL using uuidv7(). It's a time-ordered UUID format that encodes a millisecond-precision timestamp in the most significant bits, making it naturally sortable and index-friendly.
Requires PostgreSQL 18+
from fastapi_toolsets.models import UUIDv7Mixin
class Event(Base, UUIDv7Mixin):
__tablename__ = "events"
name: Mapped[str]
# id is None before flush
event = Event(name="user.signup")
session.add(event)
await session.flush()
print(event.id) # UUID('019...')
CreatedAtMixin¶
Adds a created_at: datetime column set to clock_timestamp() on insert. The column has no onupdate hook — it is intentionally immutable after the row is created.
from fastapi_toolsets.models import UUIDMixin, CreatedAtMixin
class Order(Base, UUIDMixin, CreatedAtMixin):
__tablename__ = "orders"
total: Mapped[float]
UpdatedAtMixin¶
Adds an updated_at: datetime column set to clock_timestamp() on insert and automatically updated to clock_timestamp() on every ORM-level update (via SQLAlchemy's onupdate hook).
from fastapi_toolsets.models import UUIDMixin, UpdatedAtMixin
class Post(Base, UUIDMixin, UpdatedAtMixin):
__tablename__ = "posts"
title: Mapped[str]
post = Post(title="Hello")
await session.flush()
await session.refresh(post)
post.title = "Hello World"
await session.flush()
await session.refresh(post)
print(post.updated_at)
Note
updated_at is updated by SQLAlchemy at ORM flush time. If you update rows via raw SQL (e.g. UPDATE posts SET ...), the column will not be updated automatically — use a database trigger if you need that guarantee.
TimestampMixin¶
Convenience mixin that combines CreatedAtMixin and UpdatedAtMixin. Equivalent to inheriting both.
from fastapi_toolsets.models import UUIDMixin, TimestampMixin
class Article(Base, UUIDMixin, TimestampMixin):
__tablename__ = "articles"
title: Mapped[str]
Lifecycle events¶
The event system provides lifecycle callbacks that fire after commit. If the transaction rolls back, no callback fires.
Setup¶
Event dispatch requires EventSession. Pass it as the session class when creating your session factory:
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from fastapi_toolsets.models import EventSession
engine = create_async_engine("postgresql+asyncpg://...")
SessionLocal = async_sessionmaker(engine, expire_on_commit=False, class_=EventSession)
Callbacks fire on session.commit() only — not on savepoints.
Savepoints created by get_transaction or begin_nested() do not
trigger callbacks. All events accumulated across flushes are dispatched once
when the outermost commit() is called.
Events¶
Three event types are available, each corresponding to a ModelEvent value:
| Event | Trigger |
|---|---|
ModelEvent.CREATE |
After INSERT commit |
ModelEvent.DELETE |
After DELETE commit |
ModelEvent.UPDATE |
After UPDATE commit on a watched field |
Callbacks fire only for ORM-level changes. Rows updated via raw SQL (UPDATE ... SET ...) are not detected.
Watched fields¶
Set __watched_fields__ on the model to restrict which field changes trigger UPDATE events. It must be a tuple[str, ...] — any other type raises TypeError:
| Class attribute | UPDATE behaviour |
|---|---|
__watched_fields__ = ("status", "role") |
Only fires when status or role changes |
| (not set) | Fires when any mapped field changes |
__watched_fields__ is inherited through the class hierarchy via normal Python MRO. A subclass can override it:
class Order(Base, UUIDMixin):
__watched_fields__ = ("status",)
...
class UrgentOrder(Order):
# inherits __watched_fields__ = ("status",)
...
class PriorityOrder(Order):
__watched_fields__ = ("priority",)
# overrides parent — UPDATE fires only for priority changes
...
Registering handlers¶
Register handlers with the listens_for decorator. Every callback receives three arguments: the model instance, the ModelEvent that triggered it, and a changes dict (None for CREATE and DELETE):
from fastapi_toolsets.models import ModelEvent, UUIDMixin, listens_for
class Order(Base, UUIDMixin):
__tablename__ = "orders"
__watched_fields__ = ("status",)
status: Mapped[str]
@listens_for(Order, [ModelEvent.CREATE])
async def on_order_created(order: Order, event_type: ModelEvent, changes: None):
await notify_new_order(order.id)
@listens_for(Order, [ModelEvent.DELETE])
async def on_order_deleted(order: Order, event_type: ModelEvent, changes: None):
await notify_order_cancelled(order.id)
@listens_for(Order, [ModelEvent.UPDATE])
async def on_order_updated(order: Order, event_type: ModelEvent, changes: dict):
if "status" in changes:
await notify_status_change(order.id, changes["status"])
Multiple handlers can be registered for the same model and event. Handlers registered on a parent class also fire for subclass instances.
A single handler can listen for multiple events at once. When event_types is omitted, the handler fires for all events:
@listens_for(Order, [ModelEvent.CREATE, ModelEvent.UPDATE])
async def on_order_changed(order: Order, event_type: ModelEvent, changes: dict | None):
await invalidate_cache(order.id)
@listens_for(Order) # all events
async def on_any_order_event(order: Order, event_type: ModelEvent, changes: dict | None):
await audit_log(order.id, event_type)
Field changes format¶
The changes dict maps each watched field that changed to {"old": ..., "new": ...}. Only fields that actually changed are included. For CREATE and DELETE events, changes is None:
# CREATE / DELETE → changes is None
# status changed → {"status": {"old": "pending", "new": "shipped"}}
# two fields changed → {"status": {...}, "assigned_to": {...}}
Multiple flushes in one transaction are merged: the earliest old and latest new are preserved, and on_update fires only once per commit.