---
title: "Advanced Alchemy · Litestar Plugins"
canonical_url: "https://litestar.dev/plugins/advanced-alchemy"
last_updated: "2026-08-16T03:52:42.340Z"
meta:
description: "Ready-to-go SQLAlchemy concoctions."
"og:description": "Ready-to-go SQLAlchemy concoctions."
"og:title": "Advanced Alchemy · Litestar Plugins"
"twitter:description": "Ready-to-go SQLAlchemy concoctions."
"twitter:title": "Advanced Alchemy · Litestar Plugins"
---
#  Advanced Alchemy
Ready-to-go SQLAlchemy concoctions.
[**481.8K downloads**](https://pypistats.org/packages/advanced-alchemy)• [**795 stars**](https://github.com/litestar-org/advanced-alchemy)• [**v1.11.0**](https://github.com/litestar-org/advanced-alchemy/releases)
[**cofin**](https://github.com/cofin)
| **Project** | **Status** |
| --- | --- |
| CI/CD | [](https://github.com/litestar-org/advanced-alchemy/actions/workflows/publish.yml) [](https://github.com/litestar-org/advanced-alchemy/actions/workflows/ci.yml) [](https://github.com/litestar-org/advanced-alchemy/actions/workflows/docs.yml) |
| Quality | [](https://codecov.io/github/litestar-org/advanced-alchemy) [](https://sonarcloud.io/summary/new_code?id=litestar-org_advanced-alchemy) [](https://sonarcloud.io/summary/new_code?id=litestar-org_advanced-alchemy) [](https://sonarcloud.io/summary/new_code?id=litestar-org_advanced-alchemy) [](https://sonarcloud.io/summary/new_code?id=litestar-org_advanced-alchemy) |
| Package | [](https://badge.fury.io/py/advanced-alchemy) |
| Community | [](https://discord.gg/litestar) [](https://matrix.to/#/#litestar:matrix.org) |
| Meta | [](https://github.com/litestar-org/advanced-alchemy) [](https://github.com/python/mypy) [](https://spdx.org/licenses/) [](https://github.com/astral-sh/ruff) |
Check out the [project documentation](https://advanced-alchemy.litestar.dev) 📚 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
- [SQLModel](https://sqlmodel.tiangolo.com/) compatibility — use `**SQLModel**` `**table=True**` models directly with repositories and services
- Composite primary key support — work with multi-column primary keys across repositories, services, and bulk operations
- Read/write replica routing with automatic query routing, round-robin/random replica selection, and sticky-primary mode
- Dogpile caching integration for query result caching
- Built in `**File Object**` data type for storing objects:
- Unified interface for various storage backends ([`**fsspec**`](https://filesystem-spec.readthedocs.io/en/latest/) and [`**obstore**`](https://developmentseed.org/obstore/latest/))
- 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 ([`**uuid-utils**`](https://github.com/aminalaee/uuid-utils) when installed, otherwise native in Python 3.14+, with UUID4 fallback on older versions).
- Integrated support for Nano ID using [`**fastnanoid**`](https://github.com/oliverlambson/fastnanoid) (install with the `**nanoid**` extra)
- Custom encrypted text type with multiple backend support including [`**pgcrypto**`](https://www.postgresql.org/docs/current/pgcrypto.html) for PostgreSQL and the Fernet implementation from [`**cryptography**`](https://cryptography.io/en/latest/) for other databases
- Custom password hashing type with multiple backend support including [`**Argon2**`](https://github.com/P-H-C/phc-winner-argon2), [`**Passlib**`](https://passlib.readthedocs.io/en/stable/), and [`**Pwdlib**`](https://pwdlib.readthedocs.io/en/stable/) with automatic salt generation and rehash-on-verify for transparent hash upgrades on login
- Custom TOTP shared-secret type that stores an authenticator seed encrypted at rest and verifies time-based one-time passwords via [`**pyotp**`](https://github.com/pyauth/pyotp) (install with the `**pyotp**` extra)
- Custom one-time-code type that stores email/SMS verification codes hashed in JSON with built-in expiry, single-use redemption, attempt lockout, and a secure code generator
- Pre-configured base classes with audit columns UUID or Big Integer primary keys and a [sentinel column](https://docs.sqlalchemy.org/en/20/core/connections.html#configuring-sentinel-columns).
- 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**`, `**IS NULL**`/ `**IS NOT NULL**`, and dates before and/or after.
- Tested support for multiple database backends including:
- SQLite via [aiosqlite](https://aiosqlite.omnilib.dev/en/stable/) or [sqlite](https://docs.python.org/3/library/sqlite3.html)
- Postgres via [asyncpg](https://magicstack.github.io/asyncpg/current/) or [psycopg3 (async or sync)](https://www.psycopg.org/psycopg3/)
- MySQL via [asyncmy](https://github.com/long2ice/asyncmy)
- Oracle via [oracledb (async or sync)](https://oracle.github.io/python-oracledb/) (tested on 18c and 23c)
- Google Spanner via [spanner-sqlalchemy](https://github.com/googleapis/python-spanner-sqlalchemy/)
- DuckDB via [duckdb\_engine](https://github.com/Mause/duckdb_engine)
- Microsoft SQL Server via [pyodbc](https://github.com/mkleehammer/pyodbc) or [aioodbc](https://github.com/aio-libs/aioodbc)
- CockroachDB via [sqlalchemy-cockroachdb (async or sync)](https://github.com/cockroachdb/sqlalchemy-cockroachdb)
- ...and much more
## Usage
### Installation
```
pip install advanced-alchemy
```
> !IMPORTANT
> Check out [the installation guide](https://advanced-alchemy.litestar.dev/latest/#installation) 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.get_many_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](https://github.com/litestar-org/advanced-alchemy/blob/main/examples/standalone.py)
### 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.get_many_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](https://github.com/litestar-org/advanced-alchemy/blob/main/examples/litestar/litestar_service.py)
#### 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](https://github.com/litestar-org/advanced-alchemy/blob/main/examples/flask/flask_services.py)
#### 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](https://github.com/litestar-org/advanced-alchemy/blob/main/examples/fastapi/fastapi_service.py)
#### 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)
```
## Contributing
All [Litestar Organization](https://github.com/litestar-org) projects will always be a community-centered, available for contributions of any size.
Before contributing, please review the [contribution guide](https://advanced-alchemy.litestar.dev/latest/contribution-guide.html).
If you have any questions, reach out to us on [Discord](https://discord.gg/litestar), our org-wide [GitHub discussions](https://github.com/orgs/litestar-org/discussions) page, or the [project-specific GitHub discussions page](https://github.com/litestar-org/advanced-alchemy/discussions).
---
An official [Litestar Organization](https://github.com/litestar-org) Project