SQLArgon

SQLAlchemy repository pattern and utilities

Documentation: https://asynq-io.github.io/sqlargon/

Repository: https://github.com/asynq-io/sqlargon

About

This library provides glue code to use sqlalchemy async sessions, core queries and orm models from one object which provides somewhat of repository pattern. This solution has few advantages:

• no need to pass session object to every function/method. Sessions are context-local and resolved by the repository itself
• write data access queries in one place
• no need to import insert, update, delete, select from sqlalchemy over and over again
• implicit cast of results to .scalars().all(), .one(), .mappings(), ...
• dialect-aware query builder (Postgres, SQLite, MySQL) for upserts, RETURNING and advisory locks
• your view model (e.g. FastAPI routes) does not need to know about the underlying storage. Repository class can be replaced at any moment with any object providing similar interface
• engines and routing policy are separate, so the same repository runs against one database, a primary with read replicas, or a set of shards

Installation

pip install sqlargon

or

uv add sqlargon

Optional extras: postgres, sqlite, mysql, pagination (cursor pagination), cron, opentelemetry, or standard for all of them:

pip install "sqlargon[standard]"

Usage

from typing import Sequence

import sqlalchemy as sa
from sqlalchemy.orm import Mapped, mapped_column

from sqlargon import Base, Database, SQLAlchemyRepository, set_default_database
from sqlargon.mixins import CreatedUpdatedMixin, UUIDModelMixin

set_default_database(Database(url="postgresql+asyncpg://localhost:5432/app"))


class User(UUIDModelMixin, CreatedUpdatedMixin, Base):
    name: Mapped[str] = mapped_column(sa.Unicode(255))
    last_name: Mapped[str | None] = mapped_column(sa.Unicode(255), nullable=True)


class UserRepository(SQLAlchemyRepository[User]):
    default_order_by = User.created_at.desc()

    async def get_by_name(self, name: str) -> User:
        # custom query, built with the repository's query builder
        return await self.select().filter_by(name=name).one()


user_repository = UserRepository()

The model is taken from the generic parameter, and __init__ takes no arguments — the repository resolves its database at call time (see Routing).

High level CRUD

user = await user_repository.create(name="John")
user = await user_repository.get(name="John")                # None if missing
user = await user_repository.get_or_create(name="John")
user = await user_repository.create_or_update(id=user_id, name="John")  # upsert

users = await user_repository.all()
users = await user_repository.list(User.name == "John")
count = await user_repository.count(User.name == "John")

user = await user_repository.update_one({"last_name": "Connor"}, User.id == user_id)
await user_repository.update_many({"last_name": "Connor"}, User.name == "John")

user = await user_repository.delete_one(User.id == user_id)
users = await user_repository.delete_many(User.name == "John")
await user_repository.remove(User.id == user_id)             # no results returned

Bulk operations

users = [{"name": "Alice"}, {"name": "Bob"}]

await user_repository.bulk_create(users)                       # ON CONFLICT DO NOTHING
created = await user_repository.bulk_create(users, return_results=True)
await user_repository.bulk_create_or_update(users)             # ON CONFLICT DO UPDATE
await user_repository.bulk_update(
    values=[{"name": "Alice", "last_name": "Connor"}],
    on_={"name"},
)

Conflict handling defaults to the model's primary key as index_elements and every other column in set_. Override it per repository:

from sqlargon.typing import OnConflictOptions


class UserRepository(SQLAlchemyRepository[User]):
    @property
    def on_conflict(self) -> OnConflictOptions:
        return {"index_elements": {"id"}, "set_": {"name"}, "exclude_set": {"last_name"}}

Building queries

Query methods (select, insert, upsert, update, delete, filter/where, join, load) return a repository copy carrying the statement; any other attribute is proxied to the underlying SQLAlchemy statement, so limit, order_by, group_by, ... chain as usual. Awaiting the repository executes the statement and returns a Result; the terminal helpers cast it for you:

users = await (
    user_repository.select()
    .join(Order, Order.user_id == User.id)
    .filter(User.name == "John")
    .order_by(User.created_at)
    .limit(2)
    .all()
)

user = await user_repository.select().filter(name="John").one_or_none()
name = await user_repository.select(User.name).scalar()
rows = await user_repository.select(User.id, User.name).mappings()
result = await user_repository.insert({"name": "John"}, return_results=True)

async for row in user_repository.select().stream():
    ...

Terminal methods: all(unique=False), one(), one_or_none(), first(), scalar(), scalars(), unique(), mappings(), stream(), execute().

Transactions

atomic wraps a repository method in a single session, committed on success and rolled back on error:

from sqlargon import atomic


class UserRepository(SQLAlchemyRepository[User]):
    @atomic
    async def create_users(self, names: Sequence[str]) -> None:
        for name in names:
            await self.create(name=name)

The same works on any coroutine via the database object, which also exposes named locks (database-native advisory locks where the dialect supports them):

db = Database.from_env()


@db.atomic
async def do_work() -> None: ...


@db.with_lock(key="import")
async def import_data() -> None: ...


async with db.lock("import"):
    ...

Unit of work

Repositories declared as annotations on a unit of work share one session and one transaction:

from sqlargon import SQLAlchemyUnitOfWork


class OrdersUow(SQLAlchemyUnitOfWork):
    users: UserRepository
    orders: OrderRepository


async with OrdersUow() as uow:
    user = await uow.users.create(name="John")
    await uow.orders.create(user_id=user.id)
    await uow.commit()

A unit of work never spans databases; the member database is resolved once on __aenter__ and pinned for the whole transaction.

Routing

A repository (or unit of work) without an explicit database uses the process-wide default, set with set_default_database(...) or built lazily from DATABASE_* environment variables (DATABASE_URL, DATABASE_ECHO, DATABASE_POOL_SIZE, DATABASE_READ_REPLICAS, ...):

from sqlargon import Database, DatabaseCluster

db = Database.from_env()                 # single database
cluster = DatabaseCluster.from_env()     # primary + DATABASE_READ_REPLICAS

Bind explicitly with the database class attribute or per call with using:

from sqlargon import DatabaseCluster, read_only, using

db = DatabaseCluster.with_replicas(
    "postgresql+asyncpg://primary/app",
    read_replicas=["postgresql+asyncpg://replica-1/app"],
    auto_route=True,  # SELECTs go to replicas automatically
)


class UserRepository(SQLAlchemyRepository[User]):
    database = db

    @read_only
    async def active(self) -> Sequence[User]:
        return await self.filter(is_active=True).all()


await user_repository.using("replica_0").all()
await user_repository.using(shard_key=tenant_id).create(name="John")

with using(read_only=True):
    users = await user_repository.all()

Clusters take named databases plus a router — DefaultRouter, PrimaryReplicaRouter, ModelRouter (vertical partitioning), ShardRouter (horizontal partitioning), or any object with a route(databases, context) method. See the routing docs for the full resolution order.

Pagination

Pagination is a strategy attached to a repository class; accessed on an instance it returns a paginator typed with the repository's model:

from sqlargon.pagination import PageNumberPagination


class UserRepository(SQLAlchemyRepository[User]):
    paginate = PageNumberPagination(default_page_size=25)


page = await UserRepository().filter(User.name == "John").paginate(page=2)
page.items, page.current_page, page.page_size, page.has_more

async for page in UserRepository().paginate.pages(page_size=100):
    ...

Available strategies: PageNumberPagination, TotalPageNumberPagination, LimitOffsetPagination, TotalLimitOffsetPagination and CursorPagination (keyset, requires sqlargon[pagination]).

Column types and mixins

sqlargon.types provides dialect-aware column types: GUID with GenerateUUID / GenerateUUIDV7 server defaults, Timestamp with a now() server default and JSON (orjson-serialized). sqlargon.types.pydantic adds Pydantic and ValidatedType for pydantic-validated columns. sqlargon.mixins bundles them into UUIDModelMixin, UUIDV7ModelMixin, CreatedUpdatedMixin and SoftDeleteMixin.

FastAPI

Repository and unit-of-work __init__ take no arguments, so subclasses work directly as dependencies — no routing knobs leak into the endpoint signature:

from fastapi import Depends, FastAPI

app = FastAPI()


@app.get("/users")
async def list_users(repo: UserRepository = Depends()) -> list[UserOut]:
    return await repo.all()


@app.post("/orders")
async def create_order(data: OrderIn, uow: OrdersUow = Depends()) -> OrderOut:
    async with uow:
        return await uow.orders.create(**data.model_dump())

To route a whole endpoint, attach use_context as a dependency — it applies using(...) for the span of the request:

from sqlargon import use_context


@app.get("/reports", dependencies=[Depends(use_context(read_only=True))])
async def reports(repo: UserRepository = Depends()) -> list[UserOut]:
    return await repo.all()