Interacting with repositories

Now that we’ve covered the modelling basics, we are able to create our first repository class. The repository classes include all of the standard CRUD operations as well as a few advanced features such as pagination, filtering and bulk operations.

Tip

The full code for this tutorial can be found below in the Full Code section.

Before we jump in to the code, let’s take a look at the available functions available in the the synchronous and asynchronous repositories.

Available Functions in the Repositories

Function	Category	Description
get	Selecting Data	Select a single record by primary key. Raises an exception when no record is found.
get_one	Selecting Data	Select a single record specified by the kwargs parameters. Raises an exception when no record is found.
get_one_or_none	Selecting Data	Select a single record specified by the kwargs parameters. Returns None when no record is found.
list	Selecting Data	Select a list of records specified by the kwargs parameters. Optionally it can be filtered by the included FilterTypes args.
list_and_count	Selecting Data	Select a list of records specified by the kwargs parameters. Optionally it can be filtered by the included FilterTypes args. Results are returned as a 2 value tuple that includes the rows selected and the total count of records.
get_or_create	Creating Data	Select a single record specified by the the kwargs parameters. If no record is found, one is created with the given values. There’s an optional attribute to filter on a subset of the supplied parameters and to merge updates.
create	Creating Data	Create a new record in the database.
create_many	Creating Data	Create one or more rows in the database.
update	Updating Data	Update an existing record in the database.
update_many	Updating Data	Update one or more rows in the database.
upsert	Updating Data	A single operation that updates or inserts a record based whether or not the primary key value on the model object is populated.
upsert_many	Updating Data	Updates or inserts multiple records based whether or not the primary key value on the model object is populated.
remove	Removing Data	Remove a single record from the database.
remove_many	Removing Data	Remove one or more records from the database.

Note

• All three of the bulk DML operations will leverage dialect-specific enhancements to be as efficient as possible. In addition to using efficient bulk inserts binds, the repository will optionally leverage the multi-row RETURNING support where possible. The repository will automatically detect this support from the SQLAlchemy driver, so no additional interaction is required to enable this.

• SQL engines generally have a limit to the number of elements that can be appended into an IN clause. The repository operations will automatically break lists that exceed this limit into multiple queries that are concatenated together before return. You do not need to account for this in your own code.

In the following examples, we’ll cover a few ways that you can use the repository within your applications.

Model Repository

Here we import the SQLAlchemyAsyncRepository class and create an AuthorRepository repository class. This is all that’s required to include all of the integrated repository features.

app.py

from rich import get_console
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author

Repository Context Manager

Since we’ll be using the repository outside of a Litestar application in this script, we’ll make a simple context manager to automatically handle the creation (and cleanup) of our Author repository.

The repository_factory method will do the following for us:

• Automatically create a new DB session from the SQLAlchemy configuration.

• Rollback session when any exception occurs.

• Automatically commit after function call completes.

app.py

# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:

Creating, Updating and Removing Data

To illustrate a few ways you can manipulate data in your database, we’ll go through the various CRUD operations:

Creating Data: Here’s a simple insert operation to populate our new Author table:

app.py

    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj

Updating Data: The update method will ensure any updates made to the model object are executed on the database:

app.py

    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj

Removing Data: The remove method accepts the primary key of the row you want to delete:

app.py

    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj

Now that we’ve seen how to do single-row operations, let’s look at the bulk methods we can use.

Working with Bulk Data Operations

In this section, we delve into the powerful capabilities of the repository classes for handling bulk data operations. Our example illustrates how we can efficiently manage data in our database. Specifically, we’ll use a JSON file containing information about US states and their abbreviations.

Here’s what we’re going to cover:

Fixture Data Loading

We will introduce a method for loading fixture data. Fixture data is sample data that populates your database and helps test the behavior of your application under realistic conditions. This pattern can be extended and adjusted to meet your needs.

app.py

import json
from pathlib import Path
from typing import Any


def open_fixture(fixtures_path: Path, fixture_name: str) -> Any:
    """Loads JSON file with the specified fixture name

    Args:
        fixtures_path (Path): The path to look for fixtures
        fixture_name (str): The fixture name to load.

    Raises:
        FileNotFoundError: Fixtures not found.

    Returns:
        Any: The parsed JSON data
    """
    fixture = Path(fixtures_path / f"{fixture_name}.json")
    if fixture.exists():
        with fixture.open(mode="r", encoding="utf-8") as f:
            f_data = f.read()
        return json.loads(f_data)
    raise FileNotFoundError(f"Could not find the {fixture_name} fixture")

You can review the JSON source file here:

US State Lookup JSON

You can download it: /examples/contrib/sqlalchemy/us_state_lookup.json or view below:

us_state_lookup.json

[
  {
    "name": "Alabama",
    "abbreviation": "AL"
  },
  {
    "name": "Alaska",
    "abbreviation": "AK"
  },
  {
    "name": "Arizona",
    "abbreviation": "AZ"
  },
  {
    "name": "Arkansas",
    "abbreviation": "AR"
  },
  {
    "name": "California",
    "abbreviation": "CA"
  },
  {
    "name": "Colorado",
    "abbreviation": "CO"
  },
  {
    "name": "Connecticut",
    "abbreviation": "CT"
  },
  {
    "name": "Delaware",
    "abbreviation": "DE"
  },
  {
    "name": "District Of Columbia",
    "abbreviation": "DC"
  },
  {
    "name": "Florida",
    "abbreviation": "FL"
  },
  {
    "name": "Georgia",
    "abbreviation": "GA"
  },
  {
    "name": "Guam",
    "abbreviation": "GU"
  },
  {
    "name": "Hawaii",
    "abbreviation": "HI"
  },
  {
    "name": "Idaho",
    "abbreviation": "ID"
  },
  {
    "name": "Illinois",
    "abbreviation": "IL"
  },
  {
    "name": "Indiana",
    "abbreviation": "IN"
  },
  {
    "name": "Iowa",
    "abbreviation": "IA"
  },
  {
    "name": "Kansas",
    "abbreviation": "KS"
  },
  {
    "name": "Kentucky",
    "abbreviation": "KY"
  },
  {
    "name": "Louisiana",
    "abbreviation": "LA"
  },
  {
    "name": "Maine",
    "abbreviation": "ME"
  },
  {
    "name": "Maryland",
    "abbreviation": "MD"
  },
  {
    "name": "Massachusetts",
    "abbreviation": "MA"
  },
  {
    "name": "Michigan",
    "abbreviation": "MI"
  },
  {
    "name": "Minnesota",
    "abbreviation": "MN"
  },
  {
    "name": "Mississippi",
    "abbreviation": "MS"
  },
  {
    "name": "Missouri",
    "abbreviation": "MO"
  },
  {
    "name": "Montana",
    "abbreviation": "MT"
  },
  {
    "name": "Nebraska",
    "abbreviation": "NE"
  },
  {
    "name": "Nevada",
    "abbreviation": "NV"
  },
  {
    "name": "New Hampshire",
    "abbreviation": "NH"
  },
  {
    "name": "New Jersey",
    "abbreviation": "NJ"
  },
  {
    "name": "New Mexico",
    "abbreviation": "NM"
  },
  {
    "name": "New York",
    "abbreviation": "NY"
  },
  {
    "name": "North Carolina",
    "abbreviation": "NC"
  },
  {
    "name": "North Dakota",
    "abbreviation": "ND"
  },
  {
    "name": "Ohio",
    "abbreviation": "OH"
  },
  {
    "name": "Oklahoma",
    "abbreviation": "OK"
  },
  {
    "name": "Oregon",
    "abbreviation": "OR"
  },
  {
    "name": "Palau",
    "abbreviation": "PW"
  },
  {
    "name": "Pennsylvania",
    "abbreviation": "PA"
  },
  {
    "name": "Puerto Rico",
    "abbreviation": "PR"
  },
  {
    "name": "Rhode Island",
    "abbreviation": "RI"
  },
  {
    "name": "South Carolina",
    "abbreviation": "SC"
  },
  {
    "name": "South Dakota",
    "abbreviation": "SD"
  },
  {
    "name": "Tennessee",
    "abbreviation": "TN"
  },
  {
    "name": "Texas",
    "abbreviation": "TX"
  },
  {
    "name": "Utah",
    "abbreviation": "UT"
  },
  {
    "name": "Vermont",
    "abbreviation": "VT"
  },
  {
    "name": "Virginia",
    "abbreviation": "VA"
  },
  {
    "name": "Washington",
    "abbreviation": "WA"
  },
  {
    "name": "West Virginia",
    "abbreviation": "WV"
  },
  {
    "name": "Wisconsin",
    "abbreviation": "WI"
  },
  {
    "name": "Wyoming",
    "abbreviation": "WY"
  }
]

Bulk Insert

We’ll use our fixture data to demonstrate a bulk insert operation. This operation allows you to add multiple records to your database in a single transaction, improving performance when working with larger data sets.

app.py

from rich import get_console
from sqlalchemy import create_engine
from sqlalchemy.orm import Mapped, Session, sessionmaker

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


class USState(base.UUIDBase):
    __tablename__ = "us_state_lookup"  # type: ignore[assignment]
    abbreviation: Mapped[str]
    name: Mapped[str]


class USStateRepository(repository.SQLAlchemySyncRepository[USState]):
    """US State repository."""

    model_type = USState


engine = create_engine(
    "duckdb:///:memory:",
    future=True,
)
session_factory: sessionmaker[Session] = sessionmaker(engine, expire_on_commit=False)


def run_script() -> None:
    """Load data from a fixture."""

    # Initializes the database.
    with engine.begin() as conn:
        USState.metadata.create_all(conn)

    with session_factory() as db_session:
        # 1) Load the JSON data into the US States table.
        repo = USStateRepository(session=db_session)
        fixture = open_fixture(here, USStateRepository.model_type.__tablename__)  # type: ignore
        objs = repo.add_many([USStateRepository.model_type(**raw_obj) for raw_obj in fixture])
        db_session.commit()
        console.print(f"Created {len(objs)} new objects.")

Paginated Data Selection

Next, let’s explore how to select multiple records with pagination. This functionality is useful for handling large amounts of data by breaking the data into manageable ‘pages’ or subsets. LimitOffset is one of several filter types you can use with the repository.

app.py

from litestar.repository.filters import LimitOffset


        # 2) Select paginated data and total row count.
        created_objs, total_objs = repo.list_and_count(LimitOffset(limit=10, offset=0))
        console.print(f"Selected {len(created_objs)} records out of a total of {total_objs}.")

Bulk Delete

Here we demonstrate how to perform a bulk delete operation. Just as with the bulk insert, deleting multiple records with the batch record methods is more efficient than executing row-by-row.

app.py

        # 3) Let's remove the batch of records selected.
        deleted_objs = repo.delete_many([new_obj.id for new_obj in created_objs])
        console.print(f"Removed {len(deleted_objs)} records out of a total of {total_objs}.")

Counts

Finally, we’ll demonstrate how to count the number of records remaining in the database.

app.py

        # 4) Let's count the remaining rows
        remaining_count = repo.count()
        console.print(f"Found {remaining_count} remaining records after delete.")

Now that we have demonstrated how to interact with the repository objects outside of a Litestar application, our next example will use dependency injection to add this functionality to a Controller!

Full Code

Full Code (click to toggle)

Python 3.8+

app.py

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import date, datetime
from typing import AsyncIterator
from uuid import UUID

import anyio
from rich import get_console
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author


engine = create_async_engine(
    "sqlite+aiosqlite:///test.sqlite",
    future=True,
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)


# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:
            await db_session.commit()


async def create_author() -> Author:
    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def update_author(obj: Author) -> Author:
    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def remove_author(id: UUID) -> Author:
    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def get_author_if_exists(id: UUID) -> Author | None:
    async with repository_factory() as repo:
        obj = await repo.get_one_or_none(id=id)
        if obj is not None:
            console.print(f"Found Author record for {obj.name} with primary key {obj.id}.")
        else:
            console.print(f"Could not find Author with primary key {id}.")
        return obj


async def run_script() -> None:
    """Load data from a fixture."""
    async with engine.begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)

    # 1) create a new Author record.
    console.print("1) Adding a new record")
    author = await create_author()
    author_id = author.id

    # 2) Let's update the Author record.
    console.print("2) Updating a record.")
    author.dod = datetime.strptime("1940-12-21", "%Y-%m-%d").date()
    await update_author(author)

    # 3) Let's delete the record we just created.
    console.print("3) Removing a record.")
    await remove_author(author_id)

    # 4) Let's verify the record no longer exists.
    console.print("4) Select one or none.")
    _should_be_none = await get_author_if_exists(author_id)


if __name__ == "__main__":
    anyio.run(run_script)

Python 3.9+

app.py

from __future__ import annotations

from contextlib import asynccontextmanager
from datetime import date, datetime
from collections.abc import AsyncIterator
from uuid import UUID

import anyio
from rich import get_console
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.orm import Mapped

from litestar.plugins.sqlalchemy import base, repository

console = get_console()


# the SQLAlchemy base includes a declarative model for you to use in your models.
# The `Base` class includes a `UUID` based primary key (`id`)
class Author(base.UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    dod: Mapped[date | None]


class AuthorRepository(repository.SQLAlchemyAsyncRepository[Author]):
    """Author repository."""

    model_type = Author


engine = create_async_engine(
    "sqlite+aiosqlite:///test.sqlite",
    future=True,
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)


# let's make a simple context manager as an example here.
@asynccontextmanager
async def repository_factory() -> AsyncIterator[AuthorRepository]:
    async with session_factory() as db_session:
        try:
            yield AuthorRepository(session=db_session)
        except Exception:  # noqa: BLE001
            await db_session.rollback()
        else:
            await db_session.commit()


async def create_author() -> Author:
    async with repository_factory() as repo:
        obj = await repo.add(
            Author(
                name="F. Scott Fitzgerald",
                dob=datetime.strptime("1896-09-24", "%Y-%m-%d").date(),
            )
        )
        console.print(f"Created Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def update_author(obj: Author) -> Author:
    async with repository_factory() as repo:
        obj = await repo.update(obj)
        console.print(f"Updated Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def remove_author(id: UUID) -> Author:
    async with repository_factory() as repo:
        obj = await repo.delete(id)
        console.print(f"Deleted Author record for {obj.name} with primary key {obj.id}.")
        return obj


async def get_author_if_exists(id: UUID) -> Author | None:
    async with repository_factory() as repo:
        obj = await repo.get_one_or_none(id=id)
        if obj is not None:
            console.print(f"Found Author record for {obj.name} with primary key {obj.id}.")
        else:
            console.print(f"Could not find Author with primary key {id}.")
        return obj


async def run_script() -> None:
    """Load data from a fixture."""
    async with engine.begin() as conn:
        await conn.run_sync(base.UUIDBase.metadata.create_all)

    # 1) create a new Author record.
    console.print("1) Adding a new record")
    author = await create_author()
    author_id = author.id

    # 2) Let's update the Author record.
    console.print("2) Updating a record.")
    author.dod = datetime.strptime("1940-12-21", "%Y-%m-%d").date()
    await update_author(author)

    # 3) Let's delete the record we just created.
    console.print("3) Removing a record.")
    await remove_author(author_id)

    # 4) Let's verify the record no longer exists.
    console.print("4) Select one or none.")
    _should_be_none = await get_author_if_exists(author_id)


if __name__ == "__main__":
    anyio.run(run_script)