原文
Check out the project documentation 📚 for more information.
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 Objectdata type for storing objects: - Optimized JSON types including a custom JSON type for Oracle
- Integrated support for UUID6 and UUID7 using
uuid-utils(install with theuuidextra) - Integrated support for Nano ID using
fastnanoid(install with thenanoidextra) - 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:
- ...and much more
pip install advanced-alchemyAdvanced 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": '[email protected]', 'name': 'Cody'},
{"email": '[email protected]', 'name': 'Janek'},
{"email": '[email protected]', 'name': 'Peter'},
{"email": '[email protected]', '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
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": '[email protected]', 'name': 'Cody'},
{"email": '[email protected]', 'name': 'Janek'},
{"email": '[email protected]', 'name': 'Peter'},
{"email": '[email protected]', '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.")Advanced Alchemy works with nearly all Python web frameworks. Several helpers for popular libraries are included, and additional PRs to support others are welcomed.
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 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 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
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,
)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)All Litestar Organization projects will always be a community-centered, available for contributions of any size.
Before contributing, please review the contribution guide.
If you have any questions, reach out to us on Discord, our org-wide GitHub discussions page, or the project-specific GitHub discussions page.
An official Litestar Organization Project