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 typing import AsyncGenerator, List, Optional
2
3from advanced_alchemy.extensions.litestar.plugins.init.config.asyncio import autocommit_before_send_handler
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.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
11from litestar.exceptions import ClientException, NotFoundException
12from litestar.status_codes import HTTP_409_CONFLICT
13
14
15class Base(DeclarativeBase):
16 ...
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: 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, 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: Optional[bool], 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 result.scalars().all()
53
54
55@get("/")
56async def get_list(transaction: AsyncSession, done: Optional[bool] = None) -> List[TodoItem]:
57 return await get_todo_list(done, transaction)
58
59
60@post("/")
61async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
62 transaction.add(data)
63 return data
64
65
66@put("/{item_title:str}")
67async def update_item(item_title: str, data: TodoItem, transaction: AsyncSession) -> TodoItem:
68 todo_item = await get_todo_by_title(item_title, transaction)
69 todo_item.title = data.title
70 todo_item.done = data.done
71 return todo_item
72
73
74db_config = SQLAlchemyAsyncConfig(
75 connection_string="sqlite+aiosqlite:///todo.sqlite",
76 metadata=Base.metadata,
77 create_all=True,
78 before_send_handler=autocommit_before_send_handler,
79)
80
81app = Litestar(
82 [get_list, add_item, update_item],
83 dependencies={"transaction": provide_transaction},
84 plugins=[SQLAlchemyPlugin(db_config)],
85)
1from typing import Optional
2from collections.abc import AsyncGenerator
3
4from advanced_alchemy.extensions.litestar.plugins.init.config.asyncio import autocommit_before_send_handler
5from sqlalchemy import select
6from sqlalchemy.exc import IntegrityError, NoResultFound
7from sqlalchemy.ext.asyncio import AsyncSession
8from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
9
10from litestar import Litestar, get, post, put
11from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
12from litestar.exceptions import ClientException, NotFoundException
13from litestar.status_codes import HTTP_409_CONFLICT
14
15
16class Base(DeclarativeBase):
17 ...
18
19
20class TodoItem(Base):
21 __tablename__ = "todo_items"
22
23 title: Mapped[str] = mapped_column(primary_key=True)
24 done: Mapped[bool]
25
26
27async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
28 try:
29 async with db_session.begin():
30 yield db_session
31 except IntegrityError as exc:
32 raise ClientException(
33 status_code=HTTP_409_CONFLICT,
34 detail=str(exc),
35 ) from exc
36
37
38async def get_todo_by_title(todo_name, session: AsyncSession) -> TodoItem:
39 query = select(TodoItem).where(TodoItem.title == todo_name)
40 result = await session.execute(query)
41 try:
42 return result.scalar_one()
43 except NoResultFound as e:
44 raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
45
46
47async def get_todo_list(done: Optional[bool], session: AsyncSession) -> list[TodoItem]:
48 query = select(TodoItem)
49 if done is not None:
50 query = query.where(TodoItem.done.is_(done))
51
52 result = await session.execute(query)
53 return result.scalars().all()
54
55
56@get("/")
57async def get_list(transaction: AsyncSession, done: Optional[bool] = None) -> list[TodoItem]:
58 return await get_todo_list(done, transaction)
59
60
61@post("/")
62async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
63 transaction.add(data)
64 return data
65
66
67@put("/{item_title:str}")
68async def update_item(item_title: str, data: TodoItem, transaction: AsyncSession) -> TodoItem:
69 todo_item = await get_todo_by_title(item_title, transaction)
70 todo_item.title = data.title
71 todo_item.done = data.done
72 return todo_item
73
74
75db_config = SQLAlchemyAsyncConfig(
76 connection_string="sqlite+aiosqlite:///todo.sqlite",
77 metadata=Base.metadata,
78 create_all=True,
79 before_send_handler=autocommit_before_send_handler,
80)
81
82app = Litestar(
83 [get_list, add_item, update_item],
84 dependencies={"transaction": provide_transaction},
85 plugins=[SQLAlchemyPlugin(db_config)],
86)
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
4
5class TodoItem(Base):
6 __tablename__ = "todo_items"
7
8 title: Mapped[str] = mapped_column(primary_key=True)
9 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:
1async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
2 try:
3 async with db_session.begin():
4 yield db_session
5 except IntegrityError as exc:
6 raise ClientException(
7 status_code=HTTP_409_CONFLICT,
8 detail=str(exc),
We also define a couple of utility functions, that help us to retrieve our TODO items from the database:
1async def get_todo_by_title(todo_name, session: AsyncSession) -> TodoItem:
2 query = select(TodoItem).where(TodoItem.title == todo_name)
3 result = await session.execute(query)
4 try:
5 return result.scalar_one()
6 except NoResultFound as e:
7 raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
8
9
10async def get_todo_list(done: Optional[bool], session: AsyncSession) -> List[TodoItem]:
11 query = select(TodoItem)
12 if done is not None:
13 query = query.where(TodoItem.done.is_(done))
14
15 result = await session.execute(query)
16 return result.scalars().all()
We define our route handlers, which are the interface through which TODO items can be created, retrieved and updated:
1@get("/")
2async def get_list(transaction: AsyncSession, done: Optional[bool] = None) -> List[TodoItem]:
3 return await get_todo_list(done, transaction)
4
5
6@post("/")
7async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
8 transaction.add(data)
9 return data
10
11
12@put("/{item_title:str}")
13async def update_item(item_title: str, data: TodoItem, transaction: AsyncSession) -> TodoItem:
14 todo_item = await get_todo_by_title(item_title, transaction)
15 todo_item.title = data.title
16 todo_item.done = data.done
Finally, we define our application, using the
SQLAlchemyPlugin
to configure SQLAlchemy and manage the
engine and session lifecycle, and register our transaction
dependency.
1 create_all=True,
2 before_send_handler=autocommit_before_send_handler,
3)
4
5app = Litestar(
See also