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

 1from typing import Optional
 2from collections.abc import AsyncGenerator
 3
 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.params import FromPath, FromQuery
12from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
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: 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: FromPath[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: FromQuery[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 list(result.scalars().all())
53
54
55@get("/")
56async def get_list(transaction: AsyncSession, done: FromQuery[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: FromPath[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",
79)
80
81app = Litestar(
82    [get_list, add_item, update_item],
83    dependencies={"transaction": provide_transaction},
84    plugins=[SQLAlchemyPlugin(db_config)],
85)

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

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:

 1from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 2
 3
 4class Base(DeclarativeBase): ...
 5
 6
 7class TodoItem(Base):
 8    __tablename__ = "todo_items"
 9
10    title: Mapped[str] = mapped_column(primary_key=True)

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 typing import AsyncGenerator, List, Optional
 2
 3from sqlalchemy import select
 4from sqlalchemy.exc import IntegrityError, NoResultFound
 5from sqlalchemy.ext.asyncio import AsyncSession
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, get, post, put
 9from litestar.exceptions import ClientException, NotFoundException
10from litestar.params import FromPath, FromQuery
11from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
12    done: Mapped[bool]
13
14
15async def provide_transaction(db_session: AsyncSession) -> AsyncGenerator[AsyncSession, None]:
16    try:
17        async with db_session.begin():
18            yield db_session
19    except IntegrityError as exc:
20        raise ClientException(
21            status_code=HTTP_409_CONFLICT,
22            detail=str(exc),

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

 1from typing import AsyncGenerator, List, Optional
 2
 3from sqlalchemy import select
 4from sqlalchemy.exc import IntegrityError, NoResultFound
 5from sqlalchemy.ext.asyncio import AsyncSession
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, get, post, put
 9from litestar.exceptions import ClientException, NotFoundException
10from litestar.params import FromPath, FromQuery
11from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
12        ) from exc
13
14
15async def get_todo_by_title(todo_name: FromPath[str], session: AsyncSession) -> TodoItem:
16    query = select(TodoItem).where(TodoItem.title == todo_name)
17    result = await session.execute(query)
18    try:
19        return result.scalar_one()
20    except NoResultFound as e:
21        raise NotFoundException(detail=f"TODO {todo_name!r} not found") from e
22
23
24async def get_todo_list(done: FromQuery[Optional[bool]], session: AsyncSession) -> List[TodoItem]:
25    query = select(TodoItem)
26    if done is not None:
27        query = query.where(TodoItem.done.is_(done))
28
29    result = await session.execute(query)

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

 1from typing import AsyncGenerator, List, Optional
 2
 3from sqlalchemy import select
 4from sqlalchemy.exc import IntegrityError, NoResultFound
 5from sqlalchemy.ext.asyncio import AsyncSession
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, get, post, put
 9from litestar.exceptions import ClientException, NotFoundException
10from litestar.params import FromPath, FromQuery
11from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
12    return list(result.scalars().all())
13
14
15@get("/")
16async def get_list(transaction: AsyncSession, done: FromQuery[Optional[bool]] = None) -> List[TodoItem]:
17    return await get_todo_list(done, transaction)
18
19
20@post("/")
21async def add_item(data: TodoItem, transaction: AsyncSession) -> TodoItem:
22    transaction.add(data)
23    return data
24
25
26@put("/{item_title:str}")
27async def update_item(item_title: FromPath[str], data: TodoItem, transaction: AsyncSession) -> TodoItem:
28    todo_item = await get_todo_by_title(item_title, transaction)
29    todo_item.title = data.title
30    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.

 1from typing import AsyncGenerator, List, Optional
 2
 3from sqlalchemy import select
 4from sqlalchemy.exc import IntegrityError, NoResultFound
 5from sqlalchemy.ext.asyncio import AsyncSession
 6from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 7
 8from litestar import Litestar, get, post, put
 9from litestar.exceptions import ClientException, NotFoundException
10from litestar.params import FromPath, FromQuery
11from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
12)
13
14app = Litestar(
15    [get_list, add_item, update_item],
16    dependencies={"transaction": provide_transaction},
17    plugins=[SQLAlchemyPlugin(db_config)],

See also