SQLArgon

SQLAlchemy repository pattern and utilities

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

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

Features

• Repository pattern — one object wraps async sessions, core queries and ORM models; sessions are context-local and resolved at call time, so nothing gets passed around
• High-level CRUD — create, get, get_or_create, create_or_update, all, list, count, update_one, update_many, delete_one, delete_many and remove out of the box
• Bulk operations — bulk_create, bulk_create_or_update and bulk_update with per-repository conflict handling
• Query builder — fluent, dialect-aware statements for upserts, RETURNING, advisory locks and streaming, with terminal helpers that cast results to .scalars(), .one(), .mappings(), ...
• Multi-dialect — PostgreSQL, SQLite, MySQL and MariaDB, with capability-gated SQL generation per backend
• Transactions — @atomic and database-scoped atomic() blocks, plus named advisory locks
• Unit of work — repositories declared as annotations on a unit of work share one session and one transaction
• Database routing — clusters with read replicas, shards and vertical partitioning; using(), read_only and per-request use_context
• Pagination — page-number, offset/limit and keyset cursor strategies
• Outbox — transactional outbox with a background relay and eventiq integration
• Cron — database-backed scheduler with namespaces and safe multi-instance claiming
• Column types and mixins — UUID (v4/v7), timestamp, orjson JSON and pydantic-validated columns; mixins for UUID keys, created/updated timestamps and soft delete
• Soft delete — tombstone-based deletes via SoftDeleteRepository
• Versioned models — optimistic concurrency with UUID or PostgreSQL xmin versions
• Auditable models — append-only versioned history with point-in-time reads and restore
• Vector search — embeddings with cosine, L2, dot and L1 similarity, full-text and hybrid reciprocal-rank-fusion search on PostgreSQL and SQLite
• Internationalization — multi-locale text in a JSON column or a translation table, with per-request locales and fallback chains
• FastAPI-ready — repositories and units of work work directly as dependencies
• Alembic migrations — async-first migration setup
• OpenTelemetry — optional SQLAlchemy instrumentation

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, vectors, eventiq, opentelemetry, or standard for the drivers, pagination, cron and OpenTelemetry:

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]).

Outbox

sqlargon.outbox implements the transactional outbox pattern: a write through the repository also appends a CloudEvent-shaped row to outbox_events in the same transaction, so an event can neither be lost by a rollback nor published for a row that never committed. A background relay then publishes them in write order:

from sqlargon import Base
from sqlargon.outbox import OutboxConfig, OutboxRelay, OutboxRepository


class User(UUIDModelMixin, CreatedUpdatedMixin, Base):  # is_new tells insert from update
    name: Mapped[str] = mapped_column(sa.Unicode(255))
    password: Mapped[str] = mapped_column(sa.Unicode(255))


class UserRepository(OutboxRepository[User]):
    outbox = OutboxConfig(topic="users", exclude={"password"})


await UserRepository().create(name="John", password=hashed)
# -> one outbox_events row, type "user.created", password left out

async with OutboxRelay(publish).running():  # publish is any async callable
    ...

The relay takes a plain publisher callable, so sqlargon depends on no broker client; sqlargon.integrations.eventiq adapts the rows to eventiq.CloudEvent. Only writes that go through the repository are recorded. See the outbox docs for retention, retries and ordering.

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), whose comparator carries portable JSON operators — containment and key tests, plus server-side mutation (set_key, update, remove_key) that rewrites a document in the UPDATE itself. sqlargon.types.pydantic adds Pydantic and ValidatedType for pydantic-validated columns. sqlargon.mixins bundles them into UUIDModelMixin, UUIDV7ModelMixin, CreatedUpdatedMixin and SoftDeleteMixin.

Auditable models

AuditableRepository never updates a row: every write appends the next version of the same entity, so the table is the audit log. Reads are scoped to the newest live version, so the usual methods keep their usual meaning:

from sqlargon import AuditableBase, AuditableRepository
from sqlargon.mixins import UUIDModelMixin


class Article(UUIDModelMixin, AuditableBase):
    title: Mapped[str] = mapped_column(sa.Unicode(255))


class ArticleRepository(AuditableRepository[Article]): ...


articles = ArticleRepository()

article = await articles.create(title="draft")  # version 1
await articles.update_one({"title": "final"}, Article.id == article.id)  # version 2

await articles.get(id=article.id)  # version 2
await articles.history(id=article.id)  # versions 1 and 2
await articles.get_version(1, id=article.id)  # version 1
await articles.at(yesterday).list()  # the state as of yesterday

await articles.remove(Article.id == article.id)  # appends a tombstoned version 3
await articles.restore(Article.id == article.id)  # and a live version 4

The version joins the primary key, so concurrent appends collide there rather than one silently winning, and update_if_match gives the cheaper check first. Versions are either a human-readable counter (AuditableBase) or a sortable UUIDv7 (UUIDAuditableBase), and sqlargon.audit relates other tables to one exact version or to whichever is newest. See the documentation for the full picture.

Internationalization

sqlargon.i18n keeps text in more than one locale and reads back whichever the current request wants. Register a locale getter and a fallback chain at startup, and attribute access stays a plain string in the active locale:

from sqlargon.i18n import TranslatedString, Translation, set_fallback_chain, set_locale_getter

set_locale_getter(lambda: request_locale.get())
set_fallback_chain(lambda locale: (locale or "en", "en"))


class Post(TranslationMixin, Base):
    title: Mapped[Translation] = mapped_column(TranslatedString())


await PostRepository().create(title={"en": "Hello", "pl": "Czesc"})

post = await PostRepository().select().one()
str(post.title)  # "Czesc" under a "pl" locale
post.title.data  # {"en": "Hello", "pl": "Czesc"}

Every column operator is rewritten onto the active locale's text, so Post.title == "Czesc", .like(...) and order_by need no join and no special syntax — the dialect-specific JSON read is handled per backend. Long text or many locales are better served by the second backend, translation_table, which keeps one row per locale in a side table and joins it through TranslatedRepository. See the documentation for the full picture.

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

Tests

The unit suite runs against an in-memory SQLite database and needs nothing installed:

pytest

The end-to-end suite runs the same stack against PostgreSQL, MySQL, MariaDB and an on-disk SQLite database, spinning the servers up and tearing them down with testcontainers — so it only needs a Docker daemon. It is skipped unless asked for:

pytest --e2e                              # unit suite + every backend
pytest --e2e ./tests/e2e                  # e2e only
pytest --e2e --e2e-backends=mysql,mariadb # selected backends

See tests/e2e/README.md for the layout and the per-backend capability gaps it pins down.