sqlarec

sqlarec adds a small, context-aware Active Record API on top of synchronous SQLAlchemy 2. It keeps model operations concise without forcing your application to pass a Session through every service and repository call.

It is designed for applications that want concise model operations:

user = User.query.where(User.email == "hamza@example.com").one_or_none()
users = User.query.order_by(User.name).all()
user = User.create(name="Hamza", email="hamza@example.com")

Why sqlarec

Regular SQLAlchemy makes session ownership explicit, but passing the same session through every layer can become repetitive:

def find_user(session, email):
    return session.scalars(select(User).where(User.email == email)).one_or_none()

sqlarec lets your application register a session-provider callback once. Models resolve the current session only when they execute an operation:

user = User.query.where(User.email == email).one_or_none()

This separates two responsibilities:

• Your application creates the session and decides when to commit, roll back, and close it.
• Your models use the current session without receiving it as an argument on every call.

Register the provider once in your application setup, away from model and business logic. A command runner, background-job worker, or web middleware can then create the current session and manage its transaction lifecycle. Models use that session through User.query, User.create(), or User.session without requiring every function to accept and forward a session argument.

The following framework-neutral middleware sketch shows the principle:

from contextvars import ContextVar

from sqlalchemy.orm import Session

from sqlarec import BaseModel, new_session

current_session = ContextVar[Session]("current_session")

# Register this once during application startup.
BaseModel.register_session_provider(current_session.get)


def database_middleware(handler):
    def wrapped(request):
        session = new_session()
        token = current_session.set(session)
        try:
            response = handler(request)
            session.commit()
            return response
        except Exception:
            session.rollback()
            raise
        finally:
            current_session.reset(token)
            session.close()

    return wrapped

Code executed inside that middleware can use a model from anywhere in the application:

@database_middleware
def get_user(request):
    return User.query.where(User.email == request.email).one_or_none()

The handler does not receive a session. User.query resolves the session bound by the middleware to the current execution context, so concurrent requests do not share sessions.

As a result, business code remains focused on model operations while the application retains an explicit and reliable transaction boundary. sqlarec does not depend on a web framework and never commits inside model methods.

Requirements

• Python 3.11 or later
• SQLAlchemy 2
• A synchronous SQLAlchemy Session
• uv for development

Install the package

Install the project and development tools:

uv sync

Install only runtime dependencies:

uv sync --no-dev

Create your first model

Models inherit from BaseModel and use standard SQLAlchemy mapped columns:

from sqlalchemy import Boolean, String
from sqlalchemy.orm import Mapped, mapped_column

from sqlarec import BaseModel, init_engine, new_session


class User(BaseModel):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(100))
    email: Mapped[str] = mapped_column(String(255), unique=True)
    active: Mapped[bool] = mapped_column(Boolean, default=True)


engine = init_engine("sqlite:///:memory:")
BaseModel.metadata.create_all(engine)

session = new_session()
BaseModel.register_session_provider(lambda: session)

User.create(name="Hamza", email="hamza@example.com")
session.commit()

user = User.query.one()
print(user.email)

Expected output:

hamza@example.com

BaseModel inherits from SQLAlchemy's DeclarativeBase. Relationships, constraints, indexes, and mapper configuration continue to use normal SQLAlchemy APIs.

Register the current session

Register a zero-argument callback that returns the session your application wants models to use:

from sqlarec import BaseModel


def get_session():
    return session


BaseModel.register_session_provider(get_session)

Register the provider during application startup, not separately for every concurrent request. The provider itself can retrieve a request-, job-, or context-local session:

from contextvars import ContextVar

from sqlalchemy.orm import Session

from sqlarec import BaseModel

current_session: ContextVar[Session] = ContextVar("current_session")

BaseModel.register_session_provider(current_session.get)

Middleware can set and reset this context variable around each request. This keeps concurrent request sessions isolated while allowing model calls to resolve the correct session. Only synchronous sessions are supported.

Query models and rows

Model.query and Model.select() return immutable ModelQuery wrappers. Their result methods return mapped instances:

users = User.query.all()
user = User.query.where(User.email == "hamza@example.com").one_or_none()
active = User.query.filter_by(active=True).order_by(User.name).limit(20).all()

Passing columns to Model.select() returns a RowQuery:

rows = User.select(User.id, User.email).order_by(User.id).all()
mappings = User.select(User.id, User.email).mappings().all()

The result behavior remains explicit:

User.query.all()                       -> Sequence[User]
User.select().all()                    -> Sequence[User]
User.select(User.id, User.email).all() -> Sequence[Row]

Query builders include where(), filter_by(), order_by(), group_by(), having(), join(), outerjoin(), limit(), offset(), distinct(), options(), union(), and union_all().

Create, update, and delete models

Model writes flush the current session but never commit:

user = User.create(name="Hamza", email="hamza@example.com")

user.name = "Hamza S."
user.save()

User.update().where(User.active.is_(False)).values(active=True).execute()

user.delete()
session.commit()

Keeping the transaction boundary outside model methods lets an application commit or roll back a complete unit of work atomically.

Single primary keys support direct lookup:

user = User.get_by_pk(42)
exists = User.exists(42)

String primary keys without a Python or database default receive a generated UUID hex value. Composite primary-key lookup accepts a tuple in mapper-defined key order.

Use SQLAlchemy directly when needed

Every query and update wrapper exposes its underlying SQLAlchemy statement:

query = User.query.where(User.active.is_(True))
statement = query.statement

Use the registered session for operations the wrappers do not cover:

result = User.session.execute(custom_statement)

sqlarec is an ergonomic layer, not a replacement for SQLAlchemy.

Develop the library

sqlarec/
├── src/sqlarec/
│   ├── __init__.py
│   ├── database.py
│   ├── core/
│   │   ├── base_model.py
│   │   ├── query.py
│   │   └── update.py
│   └── utils/
│       └── identifiers.py
├── tests/
├── Makefile
├── pyproject.toml
└── uv.lock

Current limitations

• You must register a session provider before model operations.
• Only synchronous SQLAlchemy sessions are supported.
• Query wrappers cover common operations; use the underlying statement for advanced SQLAlchemy features.