From e732e54518376154b79cc0337b11facd4d00f562 Mon Sep 17 00:00:00 2001 From: d3vyce <44915747+d3vyce@users.noreply.github.com> Date: Wed, 4 Mar 2026 15:14:55 +0100 Subject: [PATCH] feat: add models module (#109) * feat: add models module * docs: add models module --- docs/module/models.md | 113 +++++++++++++++ docs/reference/models.md | 22 +++ src/fastapi_toolsets/models.py | 47 +++++++ tests/test_models.py | 247 +++++++++++++++++++++++++++++++++ 4 files changed, 429 insertions(+) create mode 100644 docs/module/models.md create mode 100644 docs/reference/models.md create mode 100644 src/fastapi_toolsets/models.py create mode 100644 tests/test_models.py diff --git a/docs/module/models.md b/docs/module/models.md new file mode 100644 index 0000000..d470258 --- /dev/null +++ b/docs/module/models.md @@ -0,0 +1,113 @@ +# 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`. + +```python +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, so they are also applied when inserting rows via raw SQL outside the ORM. + +## Mixins + +### [`UUIDMixin`](../reference/models.md#fastapi_toolsets.models.UUIDMixin) + +Adds a `id: UUID` primary key generated server-side by PostgreSQL using `gen_random_uuid()` (requires PostgreSQL 13+). The value is retrieved via `RETURNING` after insert, so it is available on the Python object immediately after `flush()`. + +```python +from fastapi_toolsets.models import UUIDMixin + +class User(Base, UUIDMixin): + __tablename__ = "users" + + username: Mapped[str] + +# id is None before flush +user = User(username="alice") +await session.flush() +print(user.id) # UUID('...') +``` + +### [`CreatedAtMixin`](../reference/models.md#fastapi_toolsets.models.CreatedAtMixin) + +Adds a `created_at: datetime` column set to `NOW()` on insert. The column has no `onupdate` hook — it is intentionally immutable after the row is created. + +```python +from fastapi_toolsets.models import UUIDMixin, CreatedAtMixin + +class Order(Base, UUIDMixin, CreatedAtMixin): + __tablename__ = "orders" + + total: Mapped[float] +``` + +### [`UpdatedAtMixin`](../reference/models.md#fastapi_toolsets.models.UpdatedAtMixin) + +Adds an `updated_at: datetime` column set to `NOW()` on insert and automatically updated to `NOW()` on every ORM-level update (via SQLAlchemy's `onupdate` hook). + +```python +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`](../reference/models.md#fastapi_toolsets.models.TimestampMixin) + +Convenience mixin that combines [`CreatedAtMixin`](../reference/models.md#fastapi_toolsets.models.CreatedAtMixin) and [`UpdatedAtMixin`](../reference/models.md#fastapi_toolsets.models.UpdatedAtMixin). Equivalent to inheriting both. + +```python +from fastapi_toolsets.models import UUIDMixin, TimestampMixin + +class Article(Base, UUIDMixin, TimestampMixin): + __tablename__ = "articles" + + title: Mapped[str] +``` + +## 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. + +```python +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](../reference/models.md) diff --git a/docs/reference/models.md b/docs/reference/models.md new file mode 100644 index 0000000..e988392 --- /dev/null +++ b/docs/reference/models.md @@ -0,0 +1,22 @@ +# `models` + +Here's the reference for the SQLAlchemy model mixins provided by the `models` module. + +You can import them directly from `fastapi_toolsets.models`: + +```python +from fastapi_toolsets.models import ( + UUIDMixin, + CreatedAtMixin, + UpdatedAtMixin, + TimestampMixin, +) +``` + +## ::: fastapi_toolsets.models.UUIDMixin + +## ::: fastapi_toolsets.models.CreatedAtMixin + +## ::: fastapi_toolsets.models.UpdatedAtMixin + +## ::: fastapi_toolsets.models.TimestampMixin diff --git a/src/fastapi_toolsets/models.py b/src/fastapi_toolsets/models.py new file mode 100644 index 0000000..bbc269b --- /dev/null +++ b/src/fastapi_toolsets/models.py @@ -0,0 +1,47 @@ +"""SQLAlchemy model mixins for common column patterns.""" + +import uuid +from datetime import datetime + +from sqlalchemy import DateTime, Uuid, func, text +from sqlalchemy.orm import Mapped, mapped_column + +__all__ = [ + "UUIDMixin", + "CreatedAtMixin", + "UpdatedAtMixin", + "TimestampMixin", +] + + +class UUIDMixin: + """Mixin that adds a UUID primary key auto-generated by the database.""" + + id: Mapped[uuid.UUID] = mapped_column( + Uuid, + primary_key=True, + server_default=text("gen_random_uuid()"), + ) + + +class CreatedAtMixin: + """Mixin that adds a ``created_at`` timestamp column.""" + + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + ) + + +class UpdatedAtMixin: + """Mixin that adds an ``updated_at`` timestamp column.""" + + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + server_default=func.now(), + onupdate=func.now(), + ) + + +class TimestampMixin(CreatedAtMixin, UpdatedAtMixin): + """Mixin that combines ``created_at`` and ``updated_at`` timestamp columns.""" diff --git a/tests/test_models.py b/tests/test_models.py new file mode 100644 index 0000000..74fa5ed --- /dev/null +++ b/tests/test_models.py @@ -0,0 +1,247 @@ +"""Tests for fastapi_toolsets.models mixins.""" + +import uuid + +import pytest +from sqlalchemy import String +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column + +from fastapi_toolsets.models import ( + CreatedAtMixin, + TimestampMixin, + UUIDMixin, + UpdatedAtMixin, +) + +from .conftest import DATABASE_URL + + +class MixinBase(DeclarativeBase): + pass + + +class UUIDModel(MixinBase, UUIDMixin): + __tablename__ = "mixin_uuid_models" + + name: Mapped[str] = mapped_column(String(50)) + + +class UpdatedAtModel(MixinBase, UpdatedAtMixin): + __tablename__ = "mixin_updated_at_models" + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(50)) + + +class CreatedAtModel(MixinBase, CreatedAtMixin): + __tablename__ = "mixin_created_at_models" + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(50)) + + +class TimestampModel(MixinBase, TimestampMixin): + __tablename__ = "mixin_timestamp_models" + + id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True) + name: Mapped[str] = mapped_column(String(50)) + + +class FullMixinModel(MixinBase, UUIDMixin, UpdatedAtMixin): + __tablename__ = "mixin_full_models" + + name: Mapped[str] = mapped_column(String(50)) + + +@pytest.fixture(scope="function") +async def mixin_session(): + engine = create_async_engine(DATABASE_URL, echo=False) + async with engine.begin() as conn: + await conn.run_sync(MixinBase.metadata.create_all) + + session_factory = async_sessionmaker(engine, expire_on_commit=False) + session = session_factory() + + try: + yield session + finally: + await session.close() + async with engine.begin() as conn: + await conn.run_sync(MixinBase.metadata.drop_all) + await engine.dispose() + + +class TestUUIDMixin: + @pytest.mark.anyio + async def test_uuid_generated_by_db(self, mixin_session): + """UUID is generated server-side and populated after flush.""" + obj = UUIDModel(name="test") + mixin_session.add(obj) + await mixin_session.flush() + + assert obj.id is not None + assert isinstance(obj.id, uuid.UUID) + + @pytest.mark.anyio + async def test_uuid_is_primary_key(self): + """UUIDMixin adds id as primary key column.""" + pk_cols = [c.name for c in UUIDModel.__table__.primary_key] + assert pk_cols == ["id"] + + @pytest.mark.anyio + async def test_each_row_gets_unique_uuid(self, mixin_session): + """Each inserted row gets a distinct UUID.""" + a = UUIDModel(name="a") + b = UUIDModel(name="b") + mixin_session.add_all([a, b]) + await mixin_session.flush() + + assert a.id != b.id + + @pytest.mark.anyio + async def test_uuid_server_default_set(self): + """Column has gen_random_uuid() as server default.""" + col = UUIDModel.__table__.c["id"] + assert col.server_default is not None + assert "gen_random_uuid" in str(col.server_default.arg) + + +class TestUpdatedAtMixin: + @pytest.mark.anyio + async def test_updated_at_set_on_insert(self, mixin_session): + """updated_at is populated after insert.""" + obj = UpdatedAtModel(name="initial") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.updated_at is not None + assert obj.updated_at.tzinfo is not None # timezone-aware + + @pytest.mark.anyio + async def test_updated_at_changes_on_update(self, mixin_session): + """updated_at is updated when the row is modified.""" + obj = UpdatedAtModel(name="initial") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + original_ts = obj.updated_at + + obj.name = "modified" + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.updated_at >= original_ts + + @pytest.mark.anyio + async def test_updated_at_column_is_not_nullable(self): + """updated_at column is non-nullable.""" + col = UpdatedAtModel.__table__.c["updated_at"] + assert not col.nullable + + @pytest.mark.anyio + async def test_updated_at_has_server_default(self): + """updated_at column has a server-side default.""" + col = UpdatedAtModel.__table__.c["updated_at"] + assert col.server_default is not None + + @pytest.mark.anyio + async def test_updated_at_has_onupdate(self): + """updated_at column has an onupdate clause.""" + col = UpdatedAtModel.__table__.c["updated_at"] + assert col.onupdate is not None + + +class TestCreatedAtMixin: + @pytest.mark.anyio + async def test_created_at_set_on_insert(self, mixin_session): + """created_at is populated after insert.""" + obj = CreatedAtModel(name="new") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.created_at is not None + assert obj.created_at.tzinfo is not None # timezone-aware + + @pytest.mark.anyio + async def test_created_at_not_changed_on_update(self, mixin_session): + """created_at is not modified when the row is updated.""" + obj = CreatedAtModel(name="original") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + original_ts = obj.created_at + + obj.name = "updated" + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.created_at == original_ts + + @pytest.mark.anyio + async def test_created_at_column_is_not_nullable(self): + """created_at column is non-nullable.""" + col = CreatedAtModel.__table__.c["created_at"] + assert not col.nullable + + @pytest.mark.anyio + async def test_created_at_has_no_onupdate(self): + """created_at column has no onupdate clause.""" + col = CreatedAtModel.__table__.c["created_at"] + assert col.onupdate is None + + +class TestTimestampMixin: + @pytest.mark.anyio + async def test_both_columns_set_on_insert(self, mixin_session): + """created_at and updated_at are both populated after insert.""" + obj = TimestampModel(name="new") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.created_at is not None + assert obj.updated_at is not None + + @pytest.mark.anyio + async def test_created_at_stable_updated_at_changes_on_update(self, mixin_session): + """On update: created_at stays the same, updated_at advances.""" + obj = TimestampModel(name="original") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + original_created = obj.created_at + original_updated = obj.updated_at + + obj.name = "modified" + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert obj.created_at == original_created + assert obj.updated_at >= original_updated + + @pytest.mark.anyio + async def test_timestamp_mixin_has_both_columns(self): + """TimestampModel exposes both created_at and updated_at columns.""" + col_names = {c.name for c in TimestampModel.__table__.columns} + assert "created_at" in col_names + assert "updated_at" in col_names + + +class TestFullMixinModel: + @pytest.mark.anyio + async def test_combined_mixins_work_together(self, mixin_session): + """UUIDMixin and UpdatedAtMixin can be combined on the same model.""" + obj = FullMixinModel(name="combined") + mixin_session.add(obj) + await mixin_session.flush() + await mixin_session.refresh(obj) + + assert isinstance(obj.id, uuid.UUID) + assert obj.updated_at is not None + assert obj.updated_at.tzinfo is not None