From litestar-skills
Auto-activate for alembic/, alembic.ini, advanced_alchemy imports. Expert knowledge for Advanced Alchemy / SQLAlchemy ORM patterns. Produces ORM models with audit trails, repository/service patterns, and Alembic migrations. Use when: defining models with UUIDAuditBase, building repositories and services, configuring SQLAlchemy plugins for Litestar/FastAPI/Flask/Sanic, creating DTOs, running Alembic migrations, using custom types (EncryptedString, FileObject, PasswordHash, DateTimeUTC), composing filters and pagination, choosing base classes and mixins, configuring dogpile.cache query caching, setting up read/write replica routing, or managing file storage with obstore/fsspec backends. Not for raw SQLAlchemy without Advanced Alchemy abstractions.
npx claudepluginhub litestar-org/litestar-skills --plugin litestar-skillsThis skill uses the workspace's default tool permissions.
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:
Guides Payload CMS config (payload.config.ts), collections, fields, hooks, access control, APIs. Debugs validation errors, security, relationships, queries, transactions, hook behavior.
Builds scalable data pipelines, modern data warehouses, and real-time streaming architectures using Spark, dbt, Airflow, Kafka, and cloud platforms like Snowflake, BigQuery.
Builds production Apache Airflow DAGs with best practices for operators, sensors, testing, and deployment. For data pipelines, workflow orchestration, and batch job scheduling.
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:
id, created_at, updated_at fieldsto_model_on_create, to_model_on_update)EncryptedString, FileObject, DateTimeUTC, GUID| 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 | Purpose |
|---|---|
SQLAlchemyAsyncRepository[Model] | Standard async CRUD |
SQLAlchemyAsyncSlugRepository[Model] | CRUD + automatic slug generation |
SQLAlchemyAsyncQueryRepository | Complex read-only queries (no model_type) |
| 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.
| 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 |
SQLAlchemyAsyncRepositoryService is the primary service base class. Key behaviors:
dict to create(), update(), upsert() — the service converts via to_model_on_create / to_model_on_update lifecycle hooks before persistenceadd_many(data), update_many(data), delete_many(filters) — batched in a single transaction; prefer over calling single-row methods in a loopto_model_on_create, to_model_on_update, to_model_on_upsert — override to transform input data, hash passwords, normalize strings, etc.| 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 |
Use SQLAlchemyPlugin (composite of SQLAlchemyInitPlugin + SQLAlchemySerializationPlugin) for full integration:
SQLAlchemyPlugin: registers session provider, transaction middleware, and ORM type encoders in one callSQLAlchemyDTO: generates Litestar DTOs directly from ORM models with include/exclude field controldatetime, UUID, Decimal, Enum, and custom column typesRepositoryError, ConflictError, and NotFoundError map to HTTP 409/404 via built-in exception handlers — register with app.exception_handlers__slots__ on non-model classes, Mapped[] typing for all columnsT | None for optional fields (PEP 604 unions, never Optional[T])Repo class pattern inside service definitionsadvanced_alchemy.* imports; avoid deprecated litestar.plugins.sqlalchemy pathsfrom __future__ import annotations rule — Advanced Alchemy model modules avoid from __future__ import annotations because SQLAlchemy 2.0 Mapped[...] columns are introspected at class-creation time. Consumer application modules (handlers, services, tests) MAY and typically SHOULD use it — canonical Litestar apps use it in 100+ files.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.
Create a repository class with model_type set to your model. Use SQLAlchemyAsyncRepository for standard CRUD, SQLAlchemyAsyncSlugRepository if the model uses SlugKey.
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.
Use the framework plugin (Litestar, FastAPI, Flask, Sanic) to inject sessions and register the service as a dependency.
Run alembic revision --autogenerate -m "description" to create the migration, then review and apply with alembic upgrade head.
match_fields on services that use upsert() to avoid duplicate-key errorsschema_dump() to convert DTOs (Pydantic/msgspec/attrs) before passing to service methodsUUIDAuditBase as default base class — only deviate when you have a concrete reasonadvanced_alchemy.* imports — the old litestar.plugins.sqlalchemy paths are deprecatedfrom __future__ import annotations — SQLAlchemy 2.0 needs the real Mapped[...] type at class-creation time. Consumer modules (handlers, services, tests) MAY use it.Before delivering code, verify:
DeclarativeBase from SQLAlchemy)Mapped[] type annotationsRepo class with model_type setadvanced_alchemy.*, not deprecated pathsfrom __future__ import annotations (consumer modules may)A complete Tag entity with model, repository, and service:
"""Tag domain — model, repository, and service."""
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
Choosing between
advanced-alchemyandsqlspec:advanced-alchemy(this skill) gives you an opinionated ORM service layer withUUIDAuditBase, lifecycle hooks, repository / service / Alembic integration, andOffsetPagination[T]out of the box — pick it when you want a complete CRUD surface with attribute-style row access and you're happy inside the SQLAlchemy ecosystem.sqlspecgives you direct SQL control, 15+ driver adapters (asyncpg, oracledb, DuckDB, BigQuery, SQLite, and more), Arrow-native result streams for analytics, and a builder API when you need it — pick it when you want explicit SQL, heterogeneous database backends, or Arrow integration. Both skills integrate with Litestar via first-party plugins; see../sqlspec/SKILL.mdfor the raw-SQL / multi-adapter path.
For detailed guides and code examples, refer to the following documents in references/: