Introduction to Database Modeling and Repository Features

In this tutorial, we will cover the integrated repository features in Litestar, starting with database modeling using the included SQLAlchemy declarative model helpers. These are a series of classes and mixins that incorporate commonly used functions/column types to make working with models easier.

Tip

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

Modeling

We’ll begin by modelling the entities and relationships between authors and books. We’ll start by creating the Author table, utilizing the UUIDBase class. To keep things simple, our first model will encompass only three fields: id, name, and dob.

app.py

# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(UUIDBase):
    name: Mapped[str]

The book entity is not considered a “strong” entity and therefore always requires an author to be created. We need to configure our SQLAlchemy classes so that it is aware of this relationship. We will extend the Author model by incorporating a Book relationship. This allows each Author record to possess multiple Book records. By configuring it this way, SQLAlchemy will automatically include the necessary foreign key constraints when using the author_id field in each Book record.

app.py

# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    books: Mapped[List[Book]] = relationship(back_populates="author", lazy="selectin")
# record created, and `updated_at` is the last time the record was modified.
class Book(UUIDAuditBase):
    title: Mapped[str]
    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))

By using the audit model, we can automatically record the time a record was created and last updated.

To implement this, we will define a new Book model via the UUIDAuditBase class. Observe that the only modification here is the parent class from which we inherit. This minor change endows the book table with automatic timestamp columns (created and updated) upon deployment!

Note

If your application requires integer-based primary keys, equivalent base model and base audit model implementations can be found at BigIntBase and BigIntAuditBase respectively.

Important

Spanner only:

Using monotonically changing primary keys is considered an anti-pattern in Spanner and leads to performance problems. Additionally, Spanner does not currently include an idiom comparable to the Sequence object. This means the BigIntBase and BigIntAuditBase are not currently supported for Spanner.

Additional features provided by the built-in base models include:

• Synchronous and Asynchronous repository implementations have been tried and tested with various popular database engines. As of now, six database engines are supported: Postgres, SQLite, MySQL, DuckDB, Oracle, and Spanner.

• Automatic table name deduction from model name. For instance, a model named EventLog would correspond to the event_log database table.

• A GUID database type that establishes a native UUID in supported engines or a Binary(16) as a fallback.

• A BigInteger variant BigIntIdentity that reverts to an Integer for unsupported variants.

• A custom JsonB type that uses native JSONB where possible and Binary or Blob as an alternative.

Let’s build on this as we look at the repository classes.

Full Code

Full Code (click to expand)

app.py

from __future__ import annotations

import uuid
from datetime import date
from uuid import UUID

from sqlalchemy import ForeignKey, func, select
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
from sqlalchemy.orm import Mapped, mapped_column, relationship

from litestar import Litestar, get
from litestar.contrib.sqlalchemy.base import UUIDAuditBase, UUIDBase
from litestar.contrib.sqlalchemy.plugins import AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin


# The SQLAlchemy base includes a declarative model for you to use in your models.
# The `UUIDBase` class includes a `UUID` based primary key (`id`)
class Author(UUIDBase):
    name: Mapped[str]
    dob: Mapped[date]
    books: Mapped[list[Book]] = relationship(back_populates="author", lazy="selectin")


# The `UUIDAuditBase` class includes the same UUID` based primary key (`id`) and 2
# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
# record created, and `updated_at` is the last time the record was modified.
class Book(UUIDAuditBase):
    title: Mapped[str]
    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
    author: Mapped[Author] = relationship(lazy="joined", innerjoin=True, viewonly=True)


session_config = AsyncSessionConfig(expire_on_commit=False)
sqlalchemy_config = SQLAlchemyAsyncConfig(
    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config, create_all=True
)  # Create 'async_session' dependency.


async def on_startup() -> None:
    """Adds some dummy data if no data is present."""
    async with sqlalchemy_config.get_session() as session:
        statement = select(func.count()).select_from(Author)
        count = await session.execute(statement)
        if not count.scalar():
            author_id = uuid.uuid4()
            session.add(Author(name="Stephen King", dob=date(1954, 9, 21), id=author_id))
            session.add(Book(title="It", author_id=author_id))
            await session.commit()


@get(path="/authors")
async def get_authors(db_session: AsyncSession, db_engine: AsyncEngine) -> list[Author]:
    """Interact with SQLAlchemy engine and session."""
    return list(await db_session.scalars(select(Author)))


app = Litestar(
    route_handlers=[get_authors],
    on_startup=[on_startup],
    debug=True,
    plugins=[SQLAlchemyPlugin(config=sqlalchemy_config)],
)