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