Project Links

Meta

Author: Cody Fincher, Peter Schutt, Janek Nouvertné, Jacob Coffee

Maintainer: Litestar Developers, Cody Fincher, Jacob Coffee, Janek Nouvertné, Julien Courtes

Requires Python: >=3.9

Classifiers

Development Status

3 - Alpha

Environment

Web Environment

Intended Audience

Developers
System Administrators

License

OSI Approved :: MIT License

Natural Language

English

Operating System

OS Independent

Programming Language

Python
Python :: 3.9
Python :: 3.10
Python :: 3.11
Python :: 3.12
Python :: 3.13

Topic

Database
Database :: Database Engines/Servers
Software Development

Typing

Typed

Litestar Logo - Light

Project		Status
CI/CD
Quality
Package
Community
Meta

Advanced Alchemy

Check out the project documentation 📚 for more information.

About

A carefully crafted, thoroughly tested, optimized companion library for SQLAlchemy, offering:

Sync and async repositories, featuring common CRUD and highly optimized bulk operations
Integration with major web frameworks including Litestar, Starlette, FastAPI, Sanic
Custom-built alembic configuration and CLI with optional framework integration
Utility base classes with audit columns, primary keys and utility functions
Built in File Object data type for storing objects:
- Unified interface for various storage backends (fsspec and obstore)
- Optional lifecycle event hooks integrated with SQLAlchemy's event system to automatically save and delete files as records are inserted, updated, or deleted.
Optimized JSON types including a custom JSON type for Oracle
Integrated support for UUID6 and UUID7 using uuid-utils (install with the uuid extra)
Integrated support for Nano ID using fastnanoid (install with the nanoid extra)
Custom encrypted text type with multiple backend support including pgcrypto for PostgreSQL and the Fernet implementation from cryptography for other databases
Custom password hashing type with multiple backend support including Argon2, Passlib, and Pwdlib with automatic salt generation
Pre-configured base classes with audit columns UUID or Big Integer primary keys and a sentinel column.
Synchronous and asynchronous repositories featuring:
- Common CRUD operations for SQLAlchemy models
- Bulk inserts, updates, upserts, and deletes with dialect-specific enhancements
- Integrated counts, pagination, sorting, filtering with LIKE, IN, and dates before and/or after.
Tested support for multiple database backends including:
- SQLite via aiosqlite or sqlite
- Postgres via asyncpg or psycopg3 (async or sync)
- MySQL via asyncmy
- Oracle via oracledb (async or sync) (tested on 18c and 23c)
- Google Spanner via spanner-sqlalchemy
- DuckDB via duckdb_engine
- Microsoft SQL Server via pyodbc or aioodbc
- CockroachDB via sqlalchemy-cockroachdb (async or sync)
...and much more

Usage

Installation

pip install advanced-alchemy

[!IMPORTANT]
Check out the installation guide in our official documentation!

Repositories

Advanced Alchemy includes a set of asynchronous and synchronous repository classes for easy CRUD operations on your SQLAlchemy models.

Click to expand the example

from advanced_alchemy import base, repository, config
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, sessionmaker


class User(base.UUIDBase):
    # you can optionally override the generated table name by manually setting it.
    __tablename__ = "user_account"  # type: ignore[assignment]
    email: Mapped[str]
    name: Mapped[str]


class UserRepository(repository.SQLAlchemySyncRepository[User]):
    """User repository."""

    model_type = User


db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:", session_config=config.SyncSessionConfig(expire_on_commit=False))

# Initializes the database.
with db.get_engine().begin() as conn:
    User.metadata.create_all(conn)

with db.get_session() as db_session:
    repo = UserRepository(session=db_session)
    # 1) Create multiple users with `add_many`
    bulk_users = [
        {"email": 'cody@litestar.dev', 'name': 'Cody'},
        {"email": 'janek@litestar.dev', 'name': 'Janek'},
        {"email": 'peter@litestar.dev', 'name': 'Peter'},
        {"email": 'jacob@litestar.dev', 'name': 'Jacob'}
    ]
    objs = repo.add_many([User(**raw_user) for raw_user in bulk_users])
    db_session.commit()
    print(f"Created {len(objs)} new objects.")

    # 2) Select paginated data and total row count.  Pass additional filters as kwargs
    created_objs, total_objs = repo.list_and_count(LimitOffset(limit=10, offset=0), name="Cody")
    print(f"Selected {len(created_objs)} records out of a total of {total_objs}.")

    # 3) Let's remove the batch of records selected.
    deleted_objs = repo.delete_many([new_obj.id for new_obj in created_objs])
    print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.")

    # 4) Let's count the remaining rows
    remaining_count = repo.count()
    print(f"Found {remaining_count} remaining records after delete.")

For a full standalone example, see the sample here

Services

Advanced Alchemy includes an additional service class to make working with a repository easier. This class is designed to accept data as a dictionary or SQLAlchemy model, and it will handle the type conversions for you.

Here's the same example from above but using a service to create the data:

from advanced_alchemy import base, repository, filters, service, config
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, sessionmaker


class User(base.UUIDBase):
    # you can optionally override the generated table name by manually setting it.
    __tablename__ = "user_account"  # type: ignore[assignment]
    email: Mapped[str]
    name: Mapped[str]

class UserService(service.SQLAlchemySyncRepositoryService[User]):
    """User repository."""
    class Repo(repository.SQLAlchemySyncRepository[User]):
        """User repository."""

        model_type = User

    repository_type = Repo

db = config.SQLAlchemySyncConfig(connection_string="duckdb:///:memory:", session_config=config.SyncSessionConfig(expire_on_commit=False))

# Initializes the database.
with db.get_engine().begin() as conn:
    User.metadata.create_all(conn)

with db.get_session() as db_session:
    service = UserService(session=db_session)
    # 1) Create multiple users with `add_many`
    objs = service.create_many([
        {"email": 'cody@litestar.dev', 'name': 'Cody'},
        {"email": 'janek@litestar.dev', 'name': 'Janek'},
        {"email": 'peter@litestar.dev', 'name': 'Peter'},
        {"email": 'jacob@litestar.dev', 'name': 'Jacob'}
    ])
    print(objs)
    print(f"Created {len(objs)} new objects.")

    # 2) Select paginated data and total row count.  Pass additional filters as kwargs
    created_objs, total_objs = service.list_and_count(LimitOffset(limit=10, offset=0), name="Cody")
    print(f"Selected {len(created_objs)} records out of a total of {total_objs}.")

    # 3) Let's remove the batch of records selected.
    deleted_objs = service.delete_many([new_obj.id for new_obj in created_objs])
    print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.")

    # 4) Let's count the remaining rows
    remaining_count = service.count()
    print(f"Found {remaining_count} remaining records after delete.")

Web Frameworks

Advanced Alchemy works with nearly all Python web frameworks. Several helpers for popular libraries are included, and additional PRs to support others are welcomed.

Litestar

Advanced Alchemy is the official SQLAlchemy integration for Litestar.

In addition to installing with pip install advanced-alchemy, it can also be installed as a Litestar extra with pip install litestar[sqlalchemy].

Litestar Example

from litestar import Litestar
from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemyAsyncConfig
# alternately...
# from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin

alchemy = SQLAlchemyPlugin(
  config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),
)
app = Litestar(plugins=[alchemy])

For a full Litestar example, check here

Flask

Flask Example

from flask import Flask
from advanced_alchemy.extensions.flask import AdvancedAlchemy, SQLAlchemySyncConfig

app = Flask(__name__)
alchemy = AdvancedAlchemy(
    config=SQLAlchemySyncConfig(connection_string="duckdb:///:memory:"), app=app,
)

For a full Flask example, see here

FastAPI

FastAPI Example

from advanced_alchemy.extensions.fastapi import AdvancedAlchemy, SQLAlchemyAsyncConfig
from fastapi import FastAPI

app = FastAPI()
alchemy = AdvancedAlchemy(
    config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app,
)

For a full FastAPI example with optional CLI integration, see here

Starlette

Pre-built Example Apps

from advanced_alchemy.extensions.starlette import AdvancedAlchemy, SQLAlchemyAsyncConfig
from starlette.applications import Starlette

app = Starlette()
alchemy = AdvancedAlchemy(
    config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app,
)

Sanic

Pre-built Example Apps

from sanic import Sanic
from sanic_ext import Extend

from advanced_alchemy.extensions.sanic import AdvancedAlchemy, SQLAlchemyAsyncConfig

app = Sanic("AlchemySanicApp")
alchemy = AdvancedAlchemy(
    sqlalchemy_config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),
)
Extend.register(alchemy)