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#
1# The `Base` class includes a `UUID` based primary key (`id`)
2class Author(UUIDBase):
3    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#
1# The `Base` class includes a `UUID` based primary key (`id`)
2class Author(UUIDBase):
3    name: Mapped[str]
4    dob: Mapped[date]
5    books: Mapped[List[Book]] = relationship(back_populates="author", lazy="selectin")
6# record created, and `updated_at` is the last time the record was modified.
7class Book(UUIDAuditBase):
8    title: Mapped[str]
9    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#
 1from __future__ import annotations
 2
 3import uuid
 4from datetime import date
 5from typing import List
 6from uuid import UUID
 7
 8from sqlalchemy import ForeignKey, func, select
 9from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
10from sqlalchemy.orm import Mapped, mapped_column, relationship
11
12from litestar import Litestar, get
13from litestar.contrib.sqlalchemy.base import UUIDAuditBase, UUIDBase
14from litestar.contrib.sqlalchemy.plugins import AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin
15
16
17# the SQLAlchemy base includes a declarative model for you to use in your models.
18# The `Base` class includes a `UUID` based primary key (`id`)
19class Author(UUIDBase):
20    name: Mapped[str]
21    dob: Mapped[date]
22    books: Mapped[List[Book]] = relationship(back_populates="author", lazy="selectin")
23
24
25# The `AuditBase` class includes the same UUID` based primary key (`id`) and 2
26# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
27# record created, and `updated_at` is the last time the record was modified.
28class Book(UUIDAuditBase):
29    title: Mapped[str]
30    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
31    author: Mapped[Author] = relationship(lazy="joined", innerjoin=True, viewonly=True)
32
33
34session_config = AsyncSessionConfig(expire_on_commit=False)
35sqlalchemy_config = SQLAlchemyAsyncConfig(
36    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config, create_all=True
37)  # Create 'async_session' dependency.
38
39
40async def on_startup() -> None:
41    """Adds some dummy data if no data is present."""
42    async with sqlalchemy_config.get_session() as session:
43        statement = select(func.count()).select_from(Author)
44        count = await session.execute(statement)
45        if not count.scalar():
46            author_id = uuid.uuid4()
47            session.add(Author(name="Stephen King", dob=date(1954, 9, 21), id=author_id))
48            session.add(Book(title="It", author_id=author_id))
49            await session.commit()
50
51
52@get(path="/authors")
53async def get_authors(db_session: AsyncSession, db_engine: AsyncEngine) -> List[Author]:
54    """Interact with SQLAlchemy engine and session."""
55    return list(await db_session.scalars(select(Author)))
56
57
58app = Litestar(
59    route_handlers=[get_authors],
60    on_startup=[on_startup],
61    debug=True,
62    plugins=[SQLAlchemyPlugin(config=sqlalchemy_config)],
63)
app.py#
 1from __future__ import annotations
 2
 3import uuid
 4from datetime import date
 5from uuid import UUID
 6
 7from sqlalchemy import ForeignKey, func, select
 8from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession
 9from sqlalchemy.orm import Mapped, mapped_column, relationship
10
11from litestar import Litestar, get
12from litestar.contrib.sqlalchemy.base import UUIDAuditBase, UUIDBase
13from litestar.contrib.sqlalchemy.plugins import AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin
14
15
16# the SQLAlchemy base includes a declarative model for you to use in your models.
17# The `Base` class includes a `UUID` based primary key (`id`)
18class Author(UUIDBase):
19    name: Mapped[str]
20    dob: Mapped[date]
21    books: Mapped[list[Book]] = relationship(back_populates="author", lazy="selectin")
22
23
24# The `AuditBase` class includes the same UUID` based primary key (`id`) and 2
25# additional columns: `created_at` and `updated_at`. `created_at` is a timestamp of when the
26# record created, and `updated_at` is the last time the record was modified.
27class Book(UUIDAuditBase):
28    title: Mapped[str]
29    author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id"))
30    author: Mapped[Author] = relationship(lazy="joined", innerjoin=True, viewonly=True)
31
32
33session_config = AsyncSessionConfig(expire_on_commit=False)
34sqlalchemy_config = SQLAlchemyAsyncConfig(
35    connection_string="sqlite+aiosqlite:///test.sqlite", session_config=session_config, create_all=True
36)  # Create 'async_session' dependency.
37
38
39async def on_startup() -> None:
40    """Adds some dummy data if no data is present."""
41    async with sqlalchemy_config.get_session() as session:
42        statement = select(func.count()).select_from(Author)
43        count = await session.execute(statement)
44        if not count.scalar():
45            author_id = uuid.uuid4()
46            session.add(Author(name="Stephen King", dob=date(1954, 9, 21), id=author_id))
47            session.add(Book(title="It", author_id=author_id))
48            await session.commit()
49
50
51@get(path="/authors")
52async def get_authors(db_session: AsyncSession, db_engine: AsyncEngine) -> list[Author]:
53    """Interact with SQLAlchemy engine and session."""
54    return list(await db_session.scalars(select(Author)))
55
56
57app = Litestar(
58    route_handlers=[get_authors],
59    on_startup=[on_startup],
60    debug=True,
61    plugins=[SQLAlchemyPlugin(config=sqlalchemy_config)],
62)