Basic CRUD Repository and Functions

!!! info "Primary sources" - SQLModel create-data tutorial - SQLModel update-data tutorial - SQLModel select tutorial - SQLAlchemy AsyncSession API

??? abstract "Decision metadata" - Status: adopted - Decision level: advisory - Applies to: api-runtime, workers, tests - Last reviewed: 2026-07-26

Purpose

Show a small SQLModel CRUD layer in two forms:

independent functions for convenient standalone or composed operations;
a repository object that groups those functions behind one domain-oriented interface.

Every public operation accepts an optional AsyncSession. When omitted, reads resolve the cached session factory and own a short-lived session, while writes resolve the same factory and own a complete session-and-transaction scope. When supplied, reads borrow the session and writes borrow its already-active caller-owned transaction. The repository stores configuration and delegates to the same functions without changing those semantics.

Use the same vocabulary at every layer:

Operation	Function	Repository method	Scope when session is omitted	Missing-row result
Create	`create_widget()`	`create()`	Owned transaction	Not applicable
Read one	`get_widget()`	`get()`	Owned session	`None`
Read many	`list_widgets()`	`list()`	Owned session	Empty list
Update	`update_widget()`	`update()`	Owned transaction	`None`
Delete	`delete_widget()`	`delete()`	Owned transaction	`None`

Functions and repository methods both put domain arguments first. Database configuration and sessions are keyword-only infrastructure arguments. This keeps call sites analogous and makes ownership choices visible.

Models

Start with one table model when the application does not need distinct persistence and API schemas.

from sqlmodel import Field
from sqlmodel import SQLModel


class Widget(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str = Field(index=True)
    description: str | None = None

This reference uses direct field arguments and full-update semantics to keep the CRUD mechanics visible. Introduce separate create, update, or public schemas only when an API boundary needs different validation, field visibility, or partial-update behavior. See SQLModel integration for that larger modeling pattern.

Independent CRUD Functions

Functions are the simplest default when grouping state or behavior in an object adds no value. Each function is a complete operation boundary: it can run standalone by resolving the cached factory from database_url, or compose into a caller-owned scope through session.

from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession

from .session import session_scope
from .session import transaction_scope


async def create_widget(
    name: str,
    description: str | None = None,
    *,
    database_url: str,
    session: AsyncSession | None = None,
) -> Widget:
    async with transaction_scope(
        database_url=database_url,
        session=session,
    ) as active_session:
        widget = Widget(name=name, description=description)
        active_session.add(widget)
        await active_session.flush()
        return widget


async def get_widget(
    widget_id: int,
    *,
    database_url: str,
    session: AsyncSession | None = None,
) -> Widget | None:
    async with session_scope(
        database_url=database_url,
        session=session,
    ) as active_session:
        return await active_session.get(Widget, widget_id)


async def list_widgets(
    *,
    database_url: str,
    offset: int = 0,
    limit: int = 100,
    session: AsyncSession | None = None,
) -> list[Widget]:
    if offset < 0:
        raise ValueError("offset must be non-negative")
    if not 1 <= limit <= 100:
        raise ValueError("limit must be between 1 and 100")

    async with session_scope(
        database_url=database_url,
        session=session,
    ) as active_session:
        statement = select(Widget).order_by(Widget.id).offset(offset).limit(limit)
        return list(await active_session.scalars(statement))


async def update_widget(
    widget_id: int,
    name: str,
    description: str | None,
    *,
    database_url: str,
    session: AsyncSession | None = None,
) -> Widget | None:
    async with transaction_scope(
        database_url=database_url,
        session=session,
    ) as active_session:
        widget = await active_session.get(Widget, widget_id)
        if widget is None:
            return None

        widget.name = name
        widget.description = description
        await active_session.flush()
        return widget


async def delete_widget(
    widget_id: int,
    *,
    database_url: str,
    session: AsyncSession | None = None,
) -> Widget | None:
    async with transaction_scope(
        database_url=database_url,
        session=session,
    ) as active_session:
        widget = await active_session.get(Widget, widget_id)
        if widget is None:
            return None

        await active_session.delete(widget)
        await active_session.flush()
        return widget

Update and delete load the row through the same session that mutates it. This avoids accepting detached instances from an earlier standalone read and gives both operations an explicit None result that the application layer can map to a domain or HTTP error. Delete returns the loaded object for callers that need its values, but that object represents a row scheduled for deletion and must not be reused as persistent state. List operations validate their bounds and order by the primary key so pagination is deterministic. Add a unique tiebreaker whenever ordering by a non-unique field.

flush() sends pending writes and populates ordinary generated primary keys. It does not itself commit. For a standalone write, the surrounding owned transaction_scope() commits after the function body succeeds. For a supplied session, the caller's outer transaction retains commit and rollback ownership. Use await active_session.refresh(widget) only when the operation deliberately needs database-generated state that was not returned during the flush; an unconditional refresh adds another query.

Repository Object

A repository can provide a stable domain-facing interface when several callers need the same grouped operations. It stores repeatable database configuration, never a mutable session. Every method delegates to the analogous function and exposes the same optional-session contract.

from sqlmodel.ext.asyncio.session import AsyncSession

class WidgetRepository:
    def __init__(self, database_url: str) -> None:
        self.database_url = database_url

    async def create(
        self,
        name: str,
        description: str | None = None,
        *,
        session: AsyncSession | None = None,
    ) -> Widget:
        return await create_widget(
            name,
            description,
            database_url=self.database_url,
            session=session,
        )

    async def get(
        self,
        widget_id: int,
        *,
        session: AsyncSession | None = None,
    ) -> Widget | None:
        return await get_widget(
            widget_id,
            database_url=self.database_url,
            session=session,
        )

    async def list(
        self,
        *,
        offset: int = 0,
        limit: int = 100,
        session: AsyncSession | None = None,
    ) -> list[Widget]:
        return await list_widgets(
            database_url=self.database_url,
            offset=offset,
            limit=limit,
            session=session,
        )

    async def update(
        self,
        widget_id: int,
        name: str,
        description: str | None,
        *,
        session: AsyncSession | None = None,
    ) -> Widget | None:
        return await update_widget(
            widget_id,
            name,
            description,
            database_url=self.database_url,
            session=session,
        )

    async def delete(
        self,
        widget_id: int,
        *,
        session: AsyncSession | None = None,
    ) -> Widget | None:
        return await delete_widget(
            widget_id,
            database_url=self.database_url,
            session=session,
        )

The object is intentionally thin. Tests can construct it with a test database URL or pass a transaction-scoped test session to individual methods. A caller-provided session always wins and remains open after the method returns. A standalone operation closes its owned session before returning, so returned objects are detached; load every required scalar, deferred column, and relationship explicitly before the scope exits, and do not mutate those objects expecting persistence.

If a read participates in a later write, pass the same session and place both operations inside the explicit transaction. This avoids splitting one use case across sessions and keeps SQLAlchemy's autobegin behavior from obscuring transaction ownership. Add a repository only when its naming, shared query policy, dependency substitution, or domain boundary improves the application. Independent functions remain a valid and often clearer design.

Transaction Ownership

Compose multiple calls under one use-case transaction. At this boundary, a supplied session joins its already-active caller-owned transaction, while omitting the session creates a standalone session and transaction. Each nested CRUD write receives active_session, detects that transaction, and borrows it instead of committing independently.

The scope names describe exactly what they own: session_scope() manages session lifetime but never commits, while transaction_scope() manages a complete transaction only when it also creates the session. Both yield the name active_session because downstream CRUD code does not need to know whether the session was borrowed or owned.

from .session import transaction_scope


async def replace_widget(
    repository: WidgetRepository,
    widget_id: int,
    replacement_name: str,
    replacement_description: str | None = None,
    *,
    session: AsyncSession | None = None,
) -> Widget | None:
    async with transaction_scope(
        database_url=repository.database_url,
        session=session,
    ) as active_session:
        deleted_widget = await repository.delete(
            widget_id,
            session=active_session,
        )
        if deleted_widget is None:
            return None

        return await repository.create(
            replacement_name,
            replacement_description,
            session=active_session,
        )

If creation fails, deletion rolls back with it. For a caller-owned transaction, wrap the call in async with session.begin(): and pass that session. For a standalone use case, omit the session; the outer transaction_scope() commits on successful exit, rolls back on exception, and closes its owned session. Do not add direct commit() calls to CRUD functions or repository methods because that prevents callers from composing several operations atomically. See transaction boundaries and session management for ownership details.

Anti-Patterns

Storing one mutable AsyncSession on a long-lived repository object.
Constructing ad hoc factories or sessions instead of resolving the cached factory through the scope helpers.
Using session_scope() for an optional write, which would close an owned session without committing.
Accepting a supplied session for a write without requiring an active caller-owned transaction.
Calling commit() or rollback() directly instead of expressing ownership through transaction_scope().
Accepting unbounded list queries.
Accepting detached ORM instances for update or delete when an identifier can be resolved in the active session.
Accessing unloaded attributes after a standalone repository read has closed its owned session.

Operational Checks

Every CRUD call receives a task-local AsyncSession.
Standalone reads resolve the cached factory by database URL and close their owned session.
Standalone writes resolve the cached factory and own commit, rollback, and session cleanup through transaction_scope().
Supplied write sessions already have an active caller-owned transaction.
Each complete operation, service, or use-case boundary borrows an active transaction or owns a complete session-and-transaction scope.
List operations have pagination and deterministic ordering where required.
Update requires values for both mutable fields; passing None explicitly clears the nullable description.
Get, update, and delete use the same identifier and missing-row semantics.
Functions and repository methods use domain arguments first and keyword-only infrastructure arguments consistently.
Standalone reads load all state needed after their owned session closes.
Repository objects hold configuration or policy, never request-scoped session state.

Testing Checks

Create tests verify generated identifiers and persisted field values after commit.
Get and list tests cover found, missing, pagination, and ordering behavior.
List tests reject negative offsets and limits outside the supported range.
Update tests cover replacement of both mutable fields, including clearing the nullable description.
Update and delete tests cover missing identifiers without mutating the database.
Delete tests verify the returned row and its absence after commit.
Failure tests verify that a surrounding transaction rolls back all composed CRUD calls.
Optional-session read tests verify borrowed sessions remain open and owned sessions close without committing.
Optional-session write tests verify supplied transactions remain caller-owned and standalone transactions commit or roll back before closing.
Composition tests pass one active session through several CRUD calls and verify one atomic commit or rollback.

14 KiB Raw Blame History