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.
|