File: modeling.rst

package info (click to toggle)
python-advanced-alchemy 1.4.1-1
  • links: PTS, VCS
  • area: main
  • in suites: sid, trixie
  • size: 3,708 kB
  • sloc: python: 25,811; makefile: 162; javascript: 123; sh: 4
file content (303 lines) | stat: -rw-r--r-- 10,018 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
========
Modeling
========

Advanced Alchemy enhances SQLAlchemy's modeling capabilities with production-ready base classes, mixins, and specialized types.
This guide demonstrates modeling for a blog system with posts and tags, showcasing key features and best practices.

Base Classes
------------

Advanced Alchemy provides several base classes optimized for different use cases.  Any model can utilize these pre-defined declarative bases from sqlchemy.  Here's a brief overview of the included classes:

.. list-table:: Base Classes and Features
   :header-rows: 1
   :widths: 20 80

   * - Base Class
     - Features
   * - ``BigIntBase``
     - BIGINT primary keys for tables
   * - ``BigIntAuditBase``
     - BIGINT primary keys for tables, Automatic created_at/updated_at timestamps
   * - ``UUIDBase``
     - UUID primary keys
   * - ``UUIDv6Base``
     - UUIDv6 primary keys
   * - ``UUIDv7Base``
     - UUIDv7 primary keys
   * - ``UUIDAuditBase``
     - UUID primary keys, Automatic created_at/updated_at timestamps
   * - ``UUIDv6AuditBase``
     - UUIDv6 primary keys, Automatic created_at/updated_at timestamps
   * - ``UUIDv7AuditBase``
     - Time-sortable UUIDv7 primary keys, Automatic created_at/updated_at timestamps
   * - ``NanoIDBase``
     - URL-friendly unique identifiers, Shorter than UUIDs, collision resistant
   * - ``NanoIDAuditBase``
     - URL-friendly IDs with audit timestamps, Combines Nanoid benefits with audit trails

Mixins
-------

Additionally, Advanced Alchemy provides mixins to enhance model functionality:

.. list-table:: Available Mixins
   :header-rows: 1
   :widths: 20 80

   * - Mixin
     - Features
   * - ``SlugKey``
     - | Adds URL-friendly slug field
   * - ``AuditColumns``
     - | Automatic created_at/updated_at timestamps
       | Tracks record modifications
   * - ``UniqueMixin``
     - | Automatic Select or Create for many-to-many relationships


Basic Model Example
-------------------

Let's start with a simple blog post model:

.. code-block:: python

    import datetime
    from typing import Optional

    from advanced_alchemy.base import BigIntAuditBase
    from sqlalchemy.orm import Mapped, mapped_column

    class Post(BigIntAuditBase):
        """Blog post model with auto-incrementing ID and audit fields.

        Attributes:
            title: The post title
            content: The post content
            published: Publication status
            published_at: Timestamp of publication
            created_at: Timestamp of creation (from BigIntAuditBase)
            updated_at: Timestamp of last update (from BigIntAuditBase)
        """

        title: Mapped[str] = mapped_column(index=True)
        content: Mapped[str]
        published: Mapped[bool] = mapped_column(default=False)
        published_at: Mapped[Optional[datetime.datetime]] = mapped_column(default=None)

.. _many_to_many_relationships:

Many-to-Many Relationships
--------------------------

Let's implement a tagging system using a many-to-many relationship. This example demonstrates:
- Association table configuration
- Relationship configuration with lazy loading
- Slug key mixin
- Index creation

.. code-block:: python

    from __future__ import annotations
    from sqlalchemy import Column, ForeignKey, Table
    from sqlalchemy.orm import relationship
    from sqlalchemy.orm import Mapped, mapped_column
    from advanced_alchemy.base import BigIntAuditBase, orm_registry, SlugKey
    from typing import List

    # Association table for post-tag relationship
    post_tag = Table(
        "post_tag",
        orm_registry.metadata,
        Column("post_id", ForeignKey("post.id", ondelete="CASCADE"), primary_key=True),
        Column("tag_id", ForeignKey("tag.id", ondelete="CASCADE"), primary_key=True)
    )

    class Post(BigIntAuditBase):

        title: Mapped[str] = mapped_column(index=True)
        content: Mapped[str]
        published: Mapped[bool] = mapped_column(default=False)

        # Many-to-many relationship with tags
        tags: Mapped[List["Tag"]] = relationship(
            secondary=post_tag,
            back_populates="posts",
            lazy="selectin"
        )

    class Tag(BigIntAuditBase, SlugKey):
        """Tag model with automatic slug generation.

        The SlugKey mixin automatically adds a slug field to the model.
        """

        name: Mapped[str] = mapped_column(unique=True, index=True)
        posts: Mapped[List[Post]] = relationship(
            secondary=post_tag,
            back_populates="tags",
            viewonly=True
        )

If we want to interact with the models above, we might use something like the following:


.. code-block:: python

    from sqlalchemy.ext.asyncio import AsyncSession
    from advanced_alchemy.utils.text import slugify

    async def add_tags_to_post(
        db_session: AsyncSession,
        post: Post,
        tag_names: list[str]
    ) -> Post:
        """Add tags to a post, looking up existing tags and creating new ones if needed."""
        existing_tags = await db_session.scalars(
            select(Tag).filter(Tag.slug.in_([slugify(name) for name in tag_names]))
        )
        new_tags = [Tag(name=name, slug=slugify(name)) for name in tag_names if name not in {tag.name for tag in existing_tags}]
        post.tags.extend(new_tags + list(existing_tags))
        db_session.merge(post)
        await db_session.flush()
        return post


Fortunately, we can remove some of this logic thanks to :class:`UniqueMixin`.

.. _using_unique_mixin:

Using :class:`UniqueMixin`
--------------------------

:class:`UniqueMixin` provides automatic handling of unique constraints and merging of duplicate records. When using the mixin,
you must implement two classmethods: :meth:`unique_hash <UniqueMixin.unique_hash>` and :meth:`unique_filter <UniqueMixin.unique_hash>`. These methods enable:

- Automatic lookup of existing records
- Safe merging of duplicates
- Atomic get-or-create operations
- Configurable uniqueness criteria

Let's enhance our Tag model with :class:`UniqueMixin`:

.. code-block:: python

    from advanced_alchemy.base import BigIntAuditBase, SlugKey
    from advanced_alchemy.mixins import UniqueMixin
    from advanced_alchemy.utils.text import slugify
    from sqlalchemy.sql.elements import ColumnElement
    from typing import Hashable

    class Tag(BigIntAuditBase, SlugKey, UniqueMixin):
        """Tag model with unique name constraint and automatic slug generation.

        The UniqueMixin provides:
        - Automatic lookup of existing records
        - Safe merging of duplicates
        - Consistent slug generation
        """

        name: Mapped[str] = mapped_column(unique=True, index=True)
        posts: Mapped[list[Post]] = relationship(
            secondary=post_tag,
            back_populates="tags",
            viewonly=True
        )

        @classmethod
        def unique_hash(cls, name: str, slug: str | None = None) -> Hashable:
            """Generate a unique hash for deduplication."""
            return slugify(name)

        @classmethod
        def unique_filter(
            cls,
            name: str,
            slug: str | None = None,
        ) -> ColumnElement[bool]:
            """SQL filter for finding existing records."""
            return cls.slug == slugify(name)

We can now take advantage of :meth:`UniqueMixin.as_unique_async` to simplify the logic.

.. code-block:: python

    from sqlalchemy.ext.asyncio import AsyncSession
    from advanced_alchemy.utils.text import slugify

    async def add_tags_to_post(
        db_session: AsyncSession,
        post: Post,
        tag_names: list[str]
    ) -> Post:
        """Add tags to a post, creating new tags if needed."""
        # The UniqueMixin automatically handles:
        # 1. Looking up existing tags
        # 2. Creating new tags if needed
        # 3. Merging duplicates
        post.tags = [
          await Tag.as_unique_async(db_session, name=tag_text, slug=slugify(tag_text))
          for tag_text in tag_names
        ]
        db_session.merge(post)
        await db_session.flush()
        return post



Customizing Declarative Base
-----------------------------

In case one of the built in declarative bases do not meet your needs (or you already have your own), Advanced Alchemy already supports customizing the ``DeclarativeBase`` class.

Here's an example showing a class to generate a server-side UUID primary key for `postgres`:

.. code-block:: python

    import datetime
    from uuid import UUID, uuid4

    from advanced_alchemy.base import CommonTableAttributes, orm_registry
    from sqlalchemy import text
    from sqlalchemy.orm import (
        DeclarativeBase,
        Mapped,
        declared_attr,
        mapped_column,
        orm_insert_sentinel,
    )


    class ServerSideUUIDPrimaryKey:
        """UUID Primary Key Field Mixin."""

        id: Mapped[UUID] = mapped_column(default=uuid4, primary_key=True, server_default=text("gen_random_uuid()"))
        """UUID Primary key column."""

        # noinspection PyMethodParameters
        @declared_attr
        def _sentinel(cls) -> Mapped[int]:
            """Sentinel value required for SQLAlchemy bulk DML with UUIDs."""
            return orm_insert_sentinel(name="sa_orm_sentinel")


    class ServerSideUUIDBase(ServerSideUUIDPrimaryKey, CommonTableAttributes, DeclarativeBase):
        """Base for all SQLAlchemy declarative models with the custom UUID primary key ."""

        registry = orm_registry


    # Using ServerSideUUIDBase
    class User(ServerSideUUIDBase):
        """User model with ServerSideUUIDBase."""

        username: Mapped[str] = mapped_column(unique=True, index=True)
        email: Mapped[str] = mapped_column(unique=True)
        full_name: Mapped[str]
        is_active: Mapped[bool] = mapped_column(default=True)
        last_login: Mapped[datetime.datetime | None] = mapped_column(default=None)


With this foundation in place, let's look at the repository pattern.