Skip to main content

Ready-to-go SQLAlchemy concoctions.

Project description

Advanced Alchemy

Project Status
CI/CD Latest Release Tests And Linting Documentation Building
Quality Coverage Quality Gate Status Maintainability Rating Reliability Rating Security Rating
Community Discord
Meta Jolt Project types - Mypy License - MIT Jolt Sponsors linting - Ruff code style - Black

Check out the project documentation 📚 for more information.

About

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

  • 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

  • Optimized JSON types including a custom JSON type for Oracle.

  • 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
    • lambda_stmt when possible for improved query building performance
    • Integrated counts, pagination, sorting, filtering with LIKE, IN, and dates before and/or after.
  • Tested support for multiple database backends including:

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.

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@advanced-alchemy.dev', 'name': 'Cody'},
        {"email": 'janek@advanced-alchemy.dev', 'name': 'Janek'},
        {"email": 'peter@advanced-alchemy.dev', 'name': 'Peter'},
        {"email": 'jacob@advanced-alchemy.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.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@advanced-alchemy.dev', 'name': 'Cody'},
        {"email": 'janek@advanced-alchemy.dev', 'name': 'Janek'},
        {"email": 'peter@advanced-alchemy.dev', 'name': 'Peter'},
        {"email": 'jacob@advanced-alchemy.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 Litsetar.

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

from advanced_alchemy.extensions.litestar.plugins import SQLAlchemyPlugin
from advanced_alchemy.extensions.litestar.plugins.init.config import SQLAlchemyAsyncConfig

from litestar import Litestar

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

For a full Litestar example, check here

FastAPI

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 CRUD example, see here

Starlette

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,
)

Sanic

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)

Contributing

All Jolt 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.


Litestar Logo - Light
A Jolt Organization Project

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

advanced_alchemy-0.5.1.tar.gz (484.5 kB view hashes)

Uploaded Source

Built Distribution

advanced_alchemy-0.5.1-py3-none-any.whl (87.5 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page