Advanced Alchemy

Advanced Alchemy is an optimized companion library for SQLAlchemy designed to provide high-level abstractions for database interactions. It is particularly useful for developers building web applications who need robust, production-ready database patterns without manual boilerplate.

Key features include:

• Repositories: Both sync and async repositories providing common CRUD operations and highly optimized bulk operations (inserts, updates, upserts, deletes).
• Framework Integration: Built-in support for Litestar, Starlette, FastAPI, and Sanic.
• Advanced Data Types: Support for encrypted text, password hashing (Argon2, Passlib, Pwlib), TOTP shared-secrets, one-time codes, Nano ID, and UUID6/UUID7.
• File Object Support: A unified interface for storing objects using backends like fsspec or obstore, with lifecycle hooks to automatically manage files alongside SQLAlchemy records.
• Scalability Features: Read/write replica routing (round-robin, random, or sticky-primary) and Dogpile caching integration.
• Compatibility: Full support for SQLModel (using table=True models) and composite primary keys.
• Database Support: Tested across SQLite, Postgres, MySQL, Oracle, Google Spanner, DuckDB, MS SQL Server, and CockroachDB.

Advanced Alchemy provides custom SQLAlchemy types for common requirements like encryption, UTC datetimes, JSONB, and file storage. These types include proper Python type annotations, automatic dialect-specific implementations, and consistent behavior across different database backends.

Commonly used types include:

• DateTimeUTC: Ensures all datetime values are stored in UTC.
• EncryptedString / EncryptedText: For storing encrypted sensitive data.
• JsonB: For JSON storage.
• FileObject / StoredObject: For managing file storage via registered backends.

from datetime import datetime
from typing import Optional

from sqlalchemy.orm import Mapped, mapped_column
from advanced_alchemy.base import UUIDBase
from advanced_alchemy.types import (
    DateTimeUTC,
    EncryptedString,
    FileObject,
    JsonB,
    StoredObject,
    storages,
)

# Register a file storage backend
storages.register_backend("file:///tmp/", key="avatars")

class UserRecord(UUIDBase):
    __tablename__ = "user_records"
    created_at: Mapped[datetime] = mapped_column(DateTimeUTC)
    password: Mapped[str] = mapped_column(EncryptedString(key="secret-key"))
    preferences: Mapped[dict[str, str]] = mapped_column(JsonB)
    avatar: Mapped[Optional[FileObject]] = mapped_column(StoredObject(backend="avatars"))

By default, Advanced Alchemy implements read-your-writes consistency via a 'sticky-after-write' mechanism.

How it works:

1. A SELECT query routes to a replica.
2. An INSERT, UPDATE, or DELETE query routes to the primary.
3. Any subsequent SELECT query within the same session routes to the primary (it becomes 'sticky').
4. Once session.commit() is called, stickiness is reset, and subsequent reads return to using replicas.

Automatic Primary Routing:

• All write operations (INSERT/UPDATE/DELETE).
• SELECT statements using FOR UPDATE.
• SELECT statements occurring after a write within the same session.

To disable this behavior and allow reads to potentially see stale data from replicas immediately after a write, set sticky_after_write=False in RoutingConfig.

# Sticky-after-write behavior example
async with session_maker() as session:
    repo = UserRepository(session=session)

    # 1. Routes to replica
    users = await repo.get_many()

    # 2. Write routes to primary
    new_user = await repo.add(User(name="Alice"))

    # 3. Read now routes to primary (sticky-after-write)
    user = await repo.get(new_user.id)

    # 4. Commit resets stickiness
    await session.commit()

    # 5. Read can use replica again
    users = await repo.get_many()

Advanced Alchemy provides two methods for obtaining session instances depending on your execution context:

1. provide_session: Use this when you are within a request/response context (e.g., in guards or middleware). It attempts to retrieve a session from the request state if it exists, otherwise it creates a new one.
2. get_session: Use this when you are outside the request lifecycle (e.g., in CLI tasks, background jobs, or standalone scripts). It always returns a new instance from the session maker.

Note: provide_session requires the connection.app.state and connection.scope.

# Example: Using provide_session in a guard
async def my_guard(connection: ASGIConnection, _: BaseRouteHandler) -> None:
    db_session = alchemy_config.provide_session(connection.app.state, connection.scope)
    await db_session.execute(text("SELECT 1"))

# Example: Using get_session in a CLI task
async def _check_db_status() -> None:
    async with alchemy_config.get_session() as db_session:
        a_value = await db_session.execute(text("SELECT 1"))

The routing module enables database scalability by automatically distributing database load between a primary database and one or more read replicas.

Core Behaviors

• Automatic Routing: The system automatically routes SELECT queries to replicas and INSERT, UPDATE, or DELETE operations to the primary database.
• Sticky-After-Write: To prevent consistency issues (reading stale data immediately after an update), the system can route reads to the primary database for a period following a write operation.
• FOR UPDATE Detection: Queries using SELECT ... FOR UPDATE are automatically routed to the primary database to ensure row locking occurs on the source of truth.
• Replica Selection: When multiple replicas are configured, the system uses a selector (such as Round-robin or Random) to distribute load.
• Bind Group Routing: Beyond simple primary/replica splits, you can define and route to arbitrary bind groups (e.g., analytics or reporting).

Advanced Alchemy provides specialized filter objects in advanced_alchemy.filters to handle common query patterns more declaratively. These can be passed to repository methods like get_many.

Available Filter Constructs:

• CollectionFilter: Filters records where a column's value is (or is not) in a provided collection of values. Requires field_name and values.
• SearchFilter: Provides basic string search. Requires field_name and value. Supports ignore_case=True.
• NullFilter: Filters records where a column is NULL. Requires field_name.
• NotNullFilter: Filters records where a column is NOT NULL. Requires field_name.
• LimitOffset: Used for standard limit/offset pagination. Requires offset and limit.

from advanced_alchemy.filters import CollectionFilter, SearchFilter, NullFilter, NotNullFilter, LimitOffset

# Collection Filter
await repository.get_many(CollectionFilter(field_name="id", values=[1, 2, 3]))

# Search Filter
await repository.get_many(SearchFilter(field_name="title", value="query", ignore_case=True))

# Null/Not Null Filters
await repository.get_many(NullFilter(field_name="published_at"))
await repository.get_many(NotNullFilter(field_name="published_at"))

# Pagination
await repository.get_many_and_count(LimitOffset(offset=0, limit=20))