Files
fastapi-toolsets/docs/module/models.md
d3vyce f82225f995 feat: add WatchedFieldsMixin (#148)
* feat/add WatchedFieldsMixin and watch_fields decorator for field-change monitoring

* docs: add WatchedFieldsMixin

* feat: add on_event, on_create and on_delete

* docs: update README
2026-03-19 19:19:33 +01:00

7.3 KiB

Models

!!! info "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().

!!! warning "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

!!! info "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.

!!! warning "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]

WatchedFieldsMixin

!!! info "Added in v2.4"

WatchedFieldsMixin provides lifecycle callbacks that fire after commit — meaning the row is durably persisted when your callback runs. If the transaction rolls back, no callback fires.

Three callbacks are available, each corresponding to a ModelEvent value:

Callback Event Trigger
on_create() ModelEvent.CREATE After INSERT
on_delete() ModelEvent.DELETE After DELETE
on_update(changes) ModelEvent.UPDATE After UPDATE on a watched field

Server-side defaults (e.g. id, created_at) are fully populated in all callbacks. All callbacks support both async def and plain def. Use @watch to restrict which fields trigger on_update:

Decorator on_update behaviour
@watch("status", "role") Only fires when status or role changes
(no decorator) Fires when any mapped field changes

Option 1 — catch-all with on_event

Override on_event to handle all event types in one place. The specific methods delegate here by default:

from fastapi_toolsets.models import ModelEvent, UUIDMixin, WatchedFieldsMixin, watch

@watch("status")
class Order(Base, UUIDMixin, WatchedFieldsMixin):
    __tablename__ = "orders"

    status: Mapped[str]

    async def on_event(self, event: ModelEvent, changes: dict | None = None) -> None:
        if event == ModelEvent.CREATE:
            await notify_new_order(self.id)
        elif event == ModelEvent.DELETE:
            await notify_order_cancelled(self.id)
        elif event == ModelEvent.UPDATE:
            await notify_status_change(self.id, changes["status"])

Option 2 — targeted overrides

Override individual methods for more focused logic:

@watch("status")
class Order(Base, UUIDMixin, WatchedFieldsMixin):
    __tablename__ = "orders"

    status: Mapped[str]

    async def on_create(self) -> None:
        await notify_new_order(self.id)

    async def on_delete(self) -> None:
        await notify_order_cancelled(self.id)

    async def on_update(self, changes: dict) -> None:
        if "status" in changes:
            old = changes["status"]["old"]
            new = changes["status"]["new"]
            await notify_status_change(self.id, old, new)

Field changes format

The changes dict maps each watched field that changed to {"old": ..., "new": ...}. Only fields that actually changed are included:

# status changed   → {"status": {"old": "pending", "new": "shipped"}}
# two fields changed → {"status": {...}, "assigned_to": {...}}

!!! info "Multiple flushes in one transaction are merged: the earliest old and latest new are preserved, and on_update fires only once per commit."

!!! warning "Callbacks fire only for ORM-level changes. Rows updated via raw SQL (UPDATE ... SET ...) are not detected."

Composing mixins

All mixins can be combined in any order. The only constraint is that exactly one primary key must be defined — either via UUIDMixin or directly on the model.

from fastapi_toolsets.models import UUIDMixin, TimestampMixin

class Event(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "events"
    name: Mapped[str]

class Counter(Base, UpdatedAtMixin):
    __tablename__ = "counters"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    value: Mapped[int]

:material-api: API Reference