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

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

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.exceptions import ClientException
 7from litestar.status_codes import HTTP_409_CONFLICT
 8
 9
10async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
11    try:
12        async with db_session.begin():
13            yield db_session
14    except IntegrityError as exc:
15        raise ClientException(
16            status_code=HTTP_409_CONFLICT,
17            detail=str(exc),
18        ) from exc

 1from collections.abc import AsyncGenerator
 2
 3from sqlalchemy.exc import IntegrityError
 4from sqlalchemy.ext.asyncio import AsyncSession
 5
 6from litestar.exceptions import ClientException
 7from litestar.status_codes import HTTP_409_CONFLICT
 8
 9
10async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession]:
11    try:
12        async with db_session.begin():
13            yield db_session
14    except IntegrityError as exc:
15        raise ClientException(
16            status_code=HTTP_409_CONFLICT,
17            detail=str(exc),
18        ) 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
 4
 5
 6@get("/")
 7async def get_list(transaction: AsyncSession, done: bool | None = None) -> list[TodoItem]:
 8    return await get_todo_list(done, transaction)
 9
10
11@post("/")
12async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
13    transaction.add(data)
14    return data
15
16
17@put("/{item_title:str}")
18async def update_item(item_title: str, data: TodoItem, transaction: AsyncSession) -> TodoItem:
19    todo_item = await get_todo_by_title(item_title, transaction)
20    todo_item.title = data.title
21    todo_item.done = data.done
22    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