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.
The full code for this tutorial can be found below in the Full Code section.
1from litestar.contrib.sqlalchemy.base import UUIDBase 2class Author(UUIDBase): 3 name: Mapped[str] 4 dob: Mapped[date]
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:
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
relationship. This allows each
Author record to possess multiple
By configuring it this way, SQLAlchemy will automatically include the necessary foreign
key constraints when using the
author_id field in each
1from litestar.contrib.sqlalchemy.base import UUIDAuditBase, UUIDBase 2 books: Mapped[list["Book"]] = relationship(back_populates="author", lazy="selectin") 3class Book(UUIDAuditBase): 4 title: Mapped[str] 5 author_id: Mapped[UUID] = mapped_column(ForeignKey("author.id")) 6 author: Mapped[Author] = relationship(lazy="joined", innerjoin=True, viewonly=True)
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!
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
BigIntAuditBase are not currently supported for
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
EventLogwould correspond to the
GUIDdatabase type that establishes a native UUID in supported engines or a
Binary(16)as a fallback.
BigIntIdentitythat reverts to an
Integerfor unsupported variants.
JsonBtype that uses native
JSONBwhere possible and
Blobas an alternative.
Let’s build on this as we look at the repository classes.
Full Code (click to expand)
1from datetime import date 2from typing import TYPE_CHECKING 3from uuid import UUID 4 5from sqlalchemy import ForeignKey, select 6from sqlalchemy.orm import Mapped, mapped_column, relationship 7 8from litestar import Litestar, get 9from litestar.contrib.sqlalchemy.base import UUIDAuditBase, UUIDBase 10from litestar.contrib.sqlalchemy.plugins import AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin 11 12if TYPE_CHECKING: 13 from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession 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 36) # Create 'async_session' dependency. 37sqlalchemy_plugin = SQLAlchemyInitPlugin(config=sqlalchemy_config) 38 39 40async def on_startup() -> None: 41 """Initializes the database.""" 42 async with sqlalchemy_config.get_engine().begin() as conn: 43 await conn.run_sync(UUIDBase.metadata.create_all) 44 45 46@get(path="/authors") 47async def get_authors(db_session: "AsyncSession", db_engine: "AsyncEngine") -> list[Author]: 48 """Interact with SQLAlchemy engine and session.""" 49 return list(await db_session.scalars(select(Author))) 50 51 52app = Litestar( 53 route_handlers=[get_authors], 54 on_startup=[on_startup], 55 plugins=[SQLAlchemyInitPlugin(config=sqlalchemy_config)], 56)