Models¶
Added in v2.0
Reusable SQLAlchemy 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 the standard SQLAlchemy 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.
Because the values only increase, the default is declared with monotonic=True: SQLAlchemy inserts many new rows in a single INSERT ... RETURNING instead of one statement per row.
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]
Declarative base¶
TimezoneAwareMixin¶
Added in v6.0
A mixin for your declarative base, not for a model. Every Mapped[datetime] column of the models built on that base becomes a timezone-aware TIMESTAMPTZ, like the timestamp mixins above. Without it, SQLAlchemy maps datetime to TIMESTAMP WITHOUT TIME ZONE.
from datetime import datetime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from fastapi_toolsets.models import TimezoneAwareMixin
class Base(TimezoneAwareMixin, DeclarativeBase):
pass
class Event(Base):
__tablename__ = "events"
id: Mapped[int] = mapped_column(primary_key=True)
starts_at: Mapped[datetime] # TIMESTAMPTZ
A type_annotation_map set on your base replaces the mixin's. Merge them to keep both:
from sqlalchemy import Text
class Base(TimezoneAwareMixin, DeclarativeBase):
type_annotation_map = {
**TimezoneAwareMixin.type_annotation_map,
str: Text,
}
Switching an existing base to this mixin changes the type of its datetime columns: generate a migration for them.
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 of your Database, so every session the facade hands out dispatches callbacks:
from fastapi_toolsets.db import Database
from fastapi_toolsets.models import EventSession
db = Database("postgresql+asyncpg://...", session_class=EventSession)
That single argument covers every session the facade opens: the Depends(db) request dependency, db.session(), db.begin(), and db.lock_tables().
If you manage the session factory yourself instead of using Database, pass it to async_sessionmaker directly:
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 the outermost commit, not on savepoints.
Savepoints created by transaction or begin_nested() do not trigger callbacks. All events accumulated across flushes are dispatched once when the outermost transaction commits, whether that is an explicit await session.commit(), the exit of a top-level session.begin() block, or await trans.commit() on a transaction opened with trans = await session.begin().
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.