Advanced Alchemy

Overview

Advanced Alchemy is NOT a raw ORM — it is a service/repository layer built on top of SQLAlchemy 2.0+ with opinionated base classes, audit mixins, and deep framework integrations (Litestar, FastAPI, Flask, Sanic). It provides:

• Base models with automatic id, created_at, updated_at fields
• Repository pattern for type-safe async CRUD
• Service layer with lifecycle hooks (to_model_on_create, to_model_on_update)
• Framework plugins for automatic session/transaction management
• Custom types: EncryptedString, FileObject, DateTimeUTC, GUID
• Alembic integration for migrations via CLI

Quick Reference

Base Classes

Base Class	PK Type	Audit Columns	When to Use
UUIDAuditBase	UUID v4	created_at, updated_at	Default choice for most models
UUIDBase	UUID v4	None	Lookup tables, tags, no audit needed
UUIDv7AuditBase	UUID v7	created_at, updated_at	Time-sortable IDs (preferred over v6)
BigIntAuditBase	BigInt auto-increment	created_at, updated_at	Legacy systems, integer PKs
NanoidAuditBase	Nanoid string	created_at, updated_at	URL-friendly short IDs
DeclarativeBase	None (define yourself)	None	Full schema control

Repository Pattern

Repository	Purpose
SQLAlchemyAsyncRepository[Model]	Standard async CRUD
SQLAlchemyAsyncSlugRepository[Model]	CRUD + automatic slug generation
SQLAlchemyAsyncQueryRepository	Complex read-only queries (no model_type)

Service Layer

Service	Purpose
SQLAlchemyAsyncRepositoryService[Model]	Full CRUD with lifecycle hooks
SQLAlchemyAsyncRepositoryReadService[Model]	Read-only (list, get, count, exists)

Key lifecycle hooks: to_model_on_create, to_model_on_update, to_model_on_upsert.

Custom Types

Type	Purpose	Notes
FileObject	Object storage with lifecycle hooks	Tracks file state across session; auto-deletes on row delete via StoredObject tracker
PasswordHash	Hashed password storage	Supports Argon2, Passlib, and Pwdlib backends; hashes on assignment
EncryptedString	Transparent AES encryption at rest	Requires ENCRYPTION_KEY in config
UUID6 / UUID7	Time-sortable UUID variants	UUID7 preferred — monotonic ordering with millisecond timestamp prefix
DateTimeUTC	Timezone-aware UTC datetime	Stores as UTC; raises on naive datetimes

Service Layer

SQLAlchemyAsyncRepositoryService is the primary service base class. Key behaviors:

• Dict-to-model conversion: pass raw dict to create(), update(), upsert() — the service converts via to_model_on_create / to_model_on_update lifecycle hooks before persistence
• Bulk operations: add_many(data), update_many(data), delete_many(filters) — batched in a single transaction; prefer over calling single-row methods in a loop
• Lifecycle hooks: to_model_on_create, to_model_on_update, to_model_on_upsert — override to transform input data, hash passwords, normalize strings, etc.

Mixins

Mixin	Fields Added	When to Use
AuditMixin	created_at, updated_at, created_by, updated_by	Any model needing a full audit trail (who + when)
SlugMixin	slug (auto-generated)	URL-friendly identifiers derived from another field
UniqueMixin	get_or_create class method	Idempotent inserts for lookup/reference tables
SentinelMixin	_sentinel version column	Optimistic locking; raises ConflictError on stale writes

Litestar Integration

Use SQLAlchemyPlugin (composite of SQLAlchemyInitPlugin + SQLAlchemySerializationPlugin) for full integration:

• SQLAlchemyPlugin: registers session provider, transaction middleware, and ORM type encoders in one call
• SQLAlchemyDTO: generates Litestar DTOs directly from ORM models with include/exclude field control
• Type encoders: automatic serialization of datetime, UUID, Decimal, Enum, and custom column types
• Exception handling: RepositoryError, ConflictError, and NotFoundError map to HTTP 409/404 via built-in exception handlers — register with app.exception_handlers

Code Style

• __slots__ on non-model classes, Mapped[] typing for all columns
• T | None for optional fields (PEP 604 unions, never Optional[T])
• Full type annotations on all function signatures
• Inner Repo class pattern inside service definitions
• Prefer advanced_alchemy.* imports; avoid deprecated litestar.plugins.sqlalchemy paths

<workflow>

Workflow

Step 1: Define the Model

Choose the appropriate base class from the quick reference table. Use UUIDAuditBase unless you have a specific reason not to. Define columns with Mapped[] typing.

Step 2: Create the Repository

Create a repository class with model_type set to your model. Use SQLAlchemyAsyncRepository for standard CRUD, SQLAlchemyAsyncSlugRepository if the model uses SlugKey.

Step 3: Build the Service

Create a service class with an inner Repo class. Set match_fields for upsert logic. Add lifecycle hooks (to_model_on_create, to_model_on_update) for business logic transformations.

Step 4: Wire into Framework

Use the framework plugin (Litestar, FastAPI, Flask, Sanic) to inject sessions and register the service as a dependency.

Step 5: Generate Migration

Run alembic revision --autogenerate -m "description" to create the migration, then review and apply with alembic upgrade head.

</workflow> <guardrails>

Guardrails

• Always use the service layer for business logic — never put validation, hashing, or transformation logic directly in route handlers or repositories
• Repositories are for data access only — no business rules, no side effects beyond database operations
• Never bypass the service layer to call repository methods directly from handlers
• Always set match_fields on services that use upsert() to avoid duplicate-key errors
• Use schema_dump() to convert DTOs (Pydantic/msgspec/attrs) before passing to service methods
• Prefer UUIDAuditBase as default base class — only deviate when you have a concrete reason
• Use advanced_alchemy.* imports — the old litestar.plugins.sqlalchemy paths are deprecated

</guardrails> <validation>

Validation Checkpoint

Before delivering code, verify:

• Model inherits from an Advanced Alchemy base class (not raw DeclarativeBase from SQLAlchemy)
• All columns use Mapped[] type annotations
• Service has an inner Repo class with model_type set
• Business logic lives in service lifecycle hooks, not in route handlers
• Imports come from advanced_alchemy.*, not deprecated paths

</validation> <example>

Example

A complete Tag entity with model, repository, and service:

"""Tag domain — model, repository, and service."""
from __future__ import annotations

from advanced_alchemy.base import UUIDAuditBase
from advanced_alchemy.repository import SQLAlchemyAsyncRepository
from advanced_alchemy.service import SQLAlchemyAsyncRepositoryService
from sqlalchemy.orm import Mapped, mapped_column


class Tag(UUIDAuditBase):
    """Tag model with audit trail."""

    __tablename__ = "tag"

    name: Mapped[str] = mapped_column(unique=True)
    description: Mapped[str | None] = mapped_column(default=None)


class TagRepository(SQLAlchemyAsyncRepository[Tag]):
    """Data access for tags."""

    model_type = Tag


class TagService(SQLAlchemyAsyncRepositoryService[Tag]):
    """Business logic for tags."""

    class Repo(SQLAlchemyAsyncRepository[Tag]):
        model_type = Tag

    repository_type = Repo
    match_fields = ["name"]

    async def to_model_on_create(self, data):
        """Normalize tag name before creation."""
        if isinstance(data, dict) and "name" in data:
            data["name"] = data["name"].strip().lower()
        return data

</example>

---

References Index

For detailed guides and code examples, refer to the following documents in references/:

• Models Base classes, mixins, special types, relationships, PII tracking, and deferred loading.
• Repositories Async repository variants, configuration, slug repos, and query repos.
• Services Service layer, lifecycle hooks, composite services, filtering, and pagination.
• Litestar Plugin SQLAlchemy plugin config, DTOs, dependency injection, and session management.
• Migrations Alembic integration, CLI commands, metadata registry, and multi-database support.
• Types Complete catalog of custom column types: EncryptedString, FileObject, DateTimeUTC, GUID, PasswordHash, ColorType, and more.
• Base Classes Declarative base classes, UUID/BigInt/Nanoid variants, audit mixins, SlugKey, UniqueMixin, metadata registry, and custom base creation.
• Filters Filter system, pagination, SearchFilter, CollectionFilter, BeforeAfter, OrderBy, LimitOffset, and frontend integration patterns.
• Framework Integrations FastAPI, Flask, Starlette, and Sanic plugin setup, session management, and feature comparison across frameworks.
• Caching Dogpile.cache integration, CacheConfig, CacheManager API, automatic cache invalidation via session events, version-based list cache keys, singleflight stampede protection, and serialization.
• Read Replicas Read/write routing, RoutingConfig, engine groups, RoundRobinSelector/RandomSelector, sticky-after-write consistency, context managers for explicit routing, and RoutingAsyncSessionMaker.
• Storage (obstore) FileObject and StoredObject types, ObstoreBackend and FSSpecBackend configuration (S3, GCS, Azure, local), StorageRegistry, presigned URL generation, automatic file lifecycle via session tracker, and Pydantic integration.

---

Official References

• https://advanced-alchemy.litestar.dev/latest/
• https://advanced-alchemy.litestar.dev/latest/usage/services.html
• https://advanced-alchemy.litestar.dev/latest/usage/cli.html
• https://advanced-alchemy.litestar.dev/latest/usage/modeling/types.html
• https://advanced-alchemy.litestar.dev/latest/reference/types.html
• https://advanced-alchemy.litestar.dev/latest/changelog.html
• https://docs.litestar.dev/2/release-notes/changelog.html
• https://docs.sqlalchemy.org/en/20/orm/quickstart.html

Shared Styleguide Baseline

• Use shared styleguides for generic language/framework rules to reduce duplication in this skill.
• General Principles
• ORM and Advanced Alchemy
• Python
• Keep this skill focused on tool-specific workflows, edge cases, and integration details.