Final touches and recap

There is one more improvement that we can make to our application. Currently, we utilize both the SQLAlchemyInitPlugin and the SQLAlchemySerializationPlugin, but there is a shortcut for this configuration: the SQLAlchemyPlugin is a combination of the two, so we can simplify our configuration by using it instead.

Here is our final application:

 1from collections.abc import AsyncGenerator
 2
 3from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
 4from sqlalchemy import select
 5from sqlalchemy.exc import IntegrityError, NoResultFound
 6from sqlalchemy.ext.asyncio import AsyncSession
 7from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 8
 9from litestar import Litestar, get, post, put
10from litestar.di import NamedDependency
11from litestar.exceptions import ClientException, NotFoundException
12from litestar.params import FromPath, FromQuery
13from litestar.status_codes import HTTP_409_CONFLICT
14
15
16class Base(DeclarativeBase): ...
17
18
19class TodoItem(Base):
20    __tablename__ = "todo_items"
21
22    title: Mapped[str] = mapped_column(primary_key=True)
23    done: Mapped[bool]
24
25
26async def provide_transaction(db_session: NamedDependency[AsyncSession]) -> AsyncGenerator[AsyncSession, None]:
27    try:
28        async with db_session.begin():
29            yield db_session
30    except IntegrityError as exc:
31        raise ClientException(
32            status_code=HTTP_409_CONFLICT,
33            detail=str(exc),
34        ) from exc
35
36
37async def get_todo_by_title(todo_name: str, session: AsyncSession) -> TodoItem:
38    query = select(TodoItem).where(TodoItem.title == todo_name)
39    result = await session.execute(query)
40    try:
41        return result.scalar_one()
42    except NoResultFound as e:
43        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
44
45
46async def get_todo_list(done: bool | None, session: AsyncSession) -> list[TodoItem]:
47    query = select(TodoItem)
48    if done is not None:
49        query = query.where(TodoItem.done.is_(done))
50
51    result = await session.execute(query)
52    return list(result.scalars().all())
53
54
55@get("/")
56async def get_list(transaction: NamedDependency[AsyncSession], done: FromQuery[bool | None] = None) -> list[TodoItem]:
57    return await get_todo_list(done, transaction)
58
59
60@post("/")
61async def add_item(data: TodoItem, transaction: NamedDependency[AsyncSession]) -> TodoItem:
62    transaction.add(data)
63    return data
64
65
66@put("/{item_title:str}")
67async def update_item(
68    item_title: FromPath[str], data: TodoItem, transaction: NamedDependency[AsyncSession]
69) -> TodoItem:
70    todo_item = await get_todo_by_title(item_title, transaction)
71    todo_item.title = data.title
72    todo_item.done = data.done
73    return todo_item
74
75
76db_config = SQLAlchemyAsyncConfig(
77    connection_string="sqlite+aiosqlite:///todo.sqlite",
78    metadata=Base.metadata,
79    create_all=True,
80    before_send_handler="autocommit",
81)
82
83app = Litestar(
84    [get_list, add_item, update_item],
85    dependencies={"transaction": provide_transaction},
86    plugins=[SQLAlchemyPlugin(db_config)],
87)

 1from collections.abc import AsyncGenerator
 2
 3from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
 4from sqlalchemy import select
 5from sqlalchemy.exc import IntegrityError, NoResultFound
 6from sqlalchemy.ext.asyncio import AsyncSession
 7from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 8
 9from litestar import Litestar, get, post, put
10from litestar.di import NamedDependency
11from litestar.exceptions import ClientException, NotFoundException
12from litestar.params import FromPath, FromQuery
13from litestar.status_codes import HTTP_409_CONFLICT
14
15
16class Base(DeclarativeBase): ...
17
18
19class TodoItem(Base):
20    __tablename__ = "todo_items"
21
22    title: Mapped[str] = mapped_column(primary_key=True)
23    done: Mapped[bool]
24
25
26async def provide_transaction(db_session: NamedDependency[AsyncSession]) -> AsyncGenerator[AsyncSession]:
27    try:
28        async with db_session.begin():
29            yield db_session
30    except IntegrityError as exc:
31        raise ClientException(
32            status_code=HTTP_409_CONFLICT,
33            detail=str(exc),
34        ) from exc
35
36
37async def get_todo_by_title(todo_name: str, session: AsyncSession) -> TodoItem:
38    query = select(TodoItem).where(TodoItem.title == todo_name)
39    result = await session.execute(query)
40    try:
41        return result.scalar_one()
42    except NoResultFound as e:
43        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
44
45
46async def get_todo_list(done: bool | None, session: AsyncSession) -> list[TodoItem]:
47    query = select(TodoItem)
48    if done is not None:
49        query = query.where(TodoItem.done.is_(done))
50
51    result = await session.execute(query)
52    return list(result.scalars().all())
53
54
55@get("/")
56async def get_list(transaction: NamedDependency[AsyncSession], done: FromQuery[bool | None] = None) -> list[TodoItem]:
57    return await get_todo_list(done, transaction)
58
59
60@post("/")
61async def add_item(data: TodoItem, transaction: NamedDependency[AsyncSession]) -> TodoItem:
62    transaction.add(data)
63    return data
64
65
66@put("/{item_title:str}")
67async def update_item(
68    item_title: FromPath[str], data: TodoItem, transaction: NamedDependency[AsyncSession]
69) -> TodoItem:
70    todo_item = await get_todo_by_title(item_title, transaction)
71    todo_item.title = data.title
72    todo_item.done = data.done
73    return todo_item
74
75
76db_config = SQLAlchemyAsyncConfig(
77    connection_string="sqlite+aiosqlite:///todo.sqlite",
78    metadata=Base.metadata,
79    create_all=True,
80    before_send_handler="autocommit",
81)
82
83app = Litestar(
84    [get_list, add_item, update_item],
85    dependencies={"transaction": provide_transaction},
86    plugins=[SQLAlchemyPlugin(db_config)],
87)

Recap

In this tutorial, we have learned how to use the SQLAlchemy plugin to create a simple application that uses a database to store and retrieve data.

In the final application TodoItem is defined, representing a TODO item. It extends from the DeclarativeBase class provided by SQLAlchemy:

1class Base(DeclarativeBase): ...
2
3
4class TodoItem(Base):
5    __tablename__ = "todo_items"
6
7    title: Mapped[str] = mapped_column(primary_key=True)
8    done: Mapped[bool]

Next, we define a dependency that centralizes our database transaction management and error handling. This dependency depends on the db_session dependency, which is provided by the SQLAlchemy plugin, and is made available to our handlers via the transaction argument:

 1from collections.abc import AsyncGenerator
 2
 3from sqlalchemy.exc import IntegrityError
 4from sqlalchemy.ext.asyncio import AsyncSession
 5
 6from litestar.di import NamedDependency
 7from litestar.exceptions import ClientException
 8from litestar.status_codes import HTTP_409_CONFLICT
 9
10
11async def provide_transaction(db_session: NamedDependency[AsyncSession]) -> AsyncGenerator[AsyncSession, None]:
12    try:
13        async with db_session.begin():
14            yield db_session
15    except IntegrityError as exc:
16        raise ClientException(
17            status_code=HTTP_409_CONFLICT,
18            detail=str(exc),
19        ) from exc

 1from collections.abc import AsyncGenerator
 2
 3from sqlalchemy.exc import IntegrityError
 4from sqlalchemy.ext.asyncio import AsyncSession
 5
 6from litestar.di import NamedDependency
 7from litestar.exceptions import ClientException
 8from litestar.status_codes import HTTP_409_CONFLICT
 9
10
11async def provide_transaction(db_session: NamedDependency[AsyncSession]) -> AsyncGenerator[AsyncSession]:
12    try:
13        async with db_session.begin():
14            yield db_session
15    except IntegrityError as exc:
16        raise ClientException(
17            status_code=HTTP_409_CONFLICT,
18            detail=str(exc),
19        ) from exc

We also define a couple of utility functions, that help us to retrieve our TODO items from the database:

 1from sqlalchemy import select
 2from sqlalchemy.exc import NoResultFound
 3from sqlalchemy.ext.asyncio import AsyncSession
 4
 5from litestar.exceptions import NotFoundException
 6
 7
 8async def get_todo_by_title(todo_name: str, session: AsyncSession) -> TodoItem:
 9    query = select(TodoItem).where(TodoItem.title == todo_name)
10    result = await session.execute(query)
11    try:
12        return result.scalar_one()
13    except NoResultFound as e:
14        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
15
16
17async def get_todo_list(done: bool | None, session: AsyncSession) -> list[TodoItem]:
18    query = select(TodoItem)
19    if done is not None:
20        query = query.where(TodoItem.done.is_(done))
21
22    result = await session.execute(query)
23    return list(result.scalars().all())

We define our route handlers, which are the interface through which TODO items can be created, retrieved and updated:

 1from sqlalchemy.ext.asyncio import AsyncSession
 2
 3from litestar import get, post, put
 4from litestar.di import NamedDependency
 5from litestar.params import FromPath, FromQuery
 6
 7
 8@get("/")
 9async def get_list(transaction: NamedDependency[AsyncSession], done: FromQuery[bool | None] = None) -> list[TodoItem]:
10    return await get_todo_list(done, transaction)
11
12
13@post("/")
14async def add_item(data: TodoItem, transaction: NamedDependency[AsyncSession]) -> TodoItem:
15    transaction.add(data)
16    return data
17
18
19@put("/{item_title:str}")
20async def update_item(
21    item_title: FromPath[str], data: TodoItem, transaction: NamedDependency[AsyncSession]
22) -> TodoItem:
23    todo_item = await get_todo_by_title(item_title, transaction)
24    todo_item.title = data.title
25    todo_item.done = data.done
26    return todo_item

Finally, we define our application, using the SQLAlchemyPlugin to configure SQLAlchemy and manage the engine and session lifecycle, and register our transaction dependency.

1from advanced_alchemy.extensions.litestar import SQLAlchemyPlugin
2
3from litestar import Litestar
4
5app = Litestar(
6    [get_list, add_item, update_item],
7    dependencies={"transaction": provide_transaction},
8    plugins=[SQLAlchemyPlugin(db_config)],
9)

See also