| Project | Status | |
|---|---|---|
| CI/CD |    | |
| Quality |  | |
| Package |    | |
| Community |       | |
| Meta |       |
Check out the project documentation 📚 for more information.
A carefully crafted, thoroughly tested, optimized companion library for SQLAlchemy, offering:
uuid-utils (install with the uuid extra)fastnanoid (install with the nanoid extra)LIKE, IN, and dates before and/or after.pip install advanced-alchemy
[!IMPORTANT]
Check out the installation guide in our official documentation!
Advanced Alchemy includes a set of asynchronous and synchronous repository classes for easy CRUD operations on your SQLAlchemy models.
from advanced_alchemy.base import UUIDBase
from advanced_alchemy.filters import LimitOffset
from advanced_alchemy.repository import SQLAlchemySyncRepository
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, sessionmaker
class User(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(SQLAlchemySyncRepository[User]):
"""User repository."""
model_type = User
# use any compatible sqlalchemy engine.
engine = create_engine("duckdb:///:memory:")
session_factory = sessionmaker(engine, expire_on_commit=False)
# Initializes the database.
with engine.begin() as conn:
User.metadata.create_all(conn)
with session_factory() 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
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.
from advanced_alchemy.base import UUIDBase
from advanced_alchemy.filters import LimitOffset
from advanced_alchemy import SQLAlchemySyncRepository, SQLAlchemySyncRepositoryService
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, sessionmaker
class User(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(SQLAlchemySyncRepository[User]):
"""User repository."""
model_type = User
class UserService(SQLAlchemySyncRepositoryService[User]):
"""User repository."""
repository_type = UserRepository
# use any compatible sqlalchemy engine.
engine = create_engine("duckdb:///:memory:")
session_factory = sessionmaker(engine, expire_on_commit=False)
# Initializes the database.
with engine.begin() as conn:
User.metadata.create_all(conn)
with session_factory() 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.")
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].
from litestar import Litestar
from litestar.plugins.sqlalchemy import SQLAlchemyPlugin, SQLAlchemyAsyncConfig
# alternately...
# from advanced_alchemy.extensions.litestar.plugins import SQLAlchemyPlugin
# from advanced_alchemy.extensions.litestar.plugins.init.config import SQLAlchemyAsyncConfig
alchemy = SQLAlchemyPlugin(
config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"),
)
app = Litestar(plugins=[alchemy])
For a full Litestar example, check here
from fastapi import FastAPI
from advanced_alchemy.config import SQLAlchemyAsyncConfig
from advanced_alchemy.extensions.starlette import StarletteAdvancedAlchemy
app = FastAPI()
alchemy = StarletteAdvancedAlchemy(
config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app,
)
For a full FastAPI example, see here
from starlette.applications import Starlette
from advanced_alchemy.config import SQLAlchemyAsyncConfig
from advanced_alchemy.extensions.starlette import StarletteAdvancedAlchemy
app = Starlette()
alchemy = StarletteAdvancedAlchemy(
config=SQLAlchemyAsyncConfig(connection_string="sqlite+aiosqlite:///test.sqlite"), app=app,
)
from sanic import Sanic
from sanic_ext import Extend
from advanced_alchemy.config import SQLAlchemyAsyncConfig
from advanced_alchemy.extensions.sanic import SanicAdvancedAlchemy
app = Sanic("AlchemySanicApp")
alchemy = SanicAdvancedAlchemy(
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
FAQs
Ready-to-go SQLAlchemy concoctions.
We found that advanced-alchemy demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 3 open source maintainers collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Security News
PyPI confirms no security flaws were exploited in the Ultralytics supply chain attack and highlights improvements for safer package publishing.
Security News
Research
The Socket Research Team breaks down a malicious wrapper package that uses obfuscation to harvest credentials and exfiltrate sensitive data.
Research
Security News
Attackers used a malicious npm package typosquatting a popular ESLint plugin to steal sensitive data, execute commands, and exploit developer systems.