原文：Litestar-Org官方教程，TODO基础四章与SQLAlchemy五章，页面©2025 Litestar-Org；中文翻译、合并与静态技术注释：未完纪，2026-10-05。本文保留九章递进关系、代码变体与原始截图，未安装依赖、启动HTTP服务或操作数据库。各阶段是独立版本，不要把同名类和处理器重复拼入一个文件。
文档位于Litestar /2/系列，不能代替精确依赖锁。教程同时列出Python 3.8、3.9及3.10+注解写法；下方原代码附录保留这些版本差别。使用前应固定相容的Litestar 2、SQLAlchemy、Advanced Alchemy集成、aiosqlite和Pydantic等实际依赖，不把main分支代码混入。
一、应用基础：处理器、注解与ASGI服务器
先按原文安装 pip install 'litestar[standard]'。standard额外依赖带有Uvicorn和Litestar CLI；其他支持ASGI的服务器也可以使用，本教程采用Uvicorn。最小应用通过 @get("/") 定义异步hello_world，返回字符串Hello, world!，再把处理器列表传给Litestar实例。保存为app.py，在终端运行 litestar run，浏览器访问 http://127.0.0.1:8000/。这些命令只作展示，没有执行。
路径是应用URL中用于选择处理器的部分；装饰器get同时限定HTTP GET方法。装饰器语法相当于先定义函数，再执行 hello_world = get("/")(hello_world)。处理器必须注册到Litestar应用才会服务请求。
类型注解除了便于mypy或pyright等静态检查，也被Litestar用于运行时输入验证和响应序列化。例如返回注解str表示按字符串返回。Python本身通常不会因为类型注解就自动校验，但框架会主动读取它们。
Litestar不直接实现HTTP服务器，而通过ASGI协议与Uvicorn等服务器交互。CLI能自动识别app.py及其中的Litestar实例；开发时可用 litestar run --reload 在文件变化后重新加载，不能将开发重载方式直接当作生产部署配置。
二、读取列表：从字典到dataclass，再到查询验证
最初的TODO_LIST是模块级字典列表，每项有title字符串和done布尔值，示例三项为Start writing TODO list（完成）、???与Profit（未完成）。GET处理器返回列表，List[Dict[str, Union[str, bool]]]等返回注解让Litestar将结果序列化为JSON。
接着用 @dataclass 定义TodoItem(title: str, done: bool)，列表中存放数据类实例。代码更有结构，返回结果仍被自动序列化成相同JSON。Litestar也支持TypedDict、NamedTuple、Pydantic模型和attrs类，具体能力应核对安装版本。
手工读取与验证查询参数
处理器的done参数用FromQuery[str]标记后，框架从URL查询串提取同名值并传入函数。第一版只检查done是否为字符串"1"：相等时返回已完成项，否则返回未完成项。访问?done=1与?done=0看似正确，但?done=john也会走未完成分支，这暴露出验证缺口。
改进版分别接受"1"、"0"，其他值抛出HTTPException，状态码400，并给出包含原值的错误说明。异常告诉Litestar返回错误响应，而不是普通列表。
交给FromQuery[bool]转换
进一步把注解改为FromQuery[bool]，框架会先进行HTTP布尔值转换与校验，然后处理器只需比较item.done == done。true和1转换为True，false和0转换为False；这不是调用Python的bool(raw_string)，后者会把非空字符串"john"当成True。无效值由框架返回描述性的400错误。
![FromQuery[bool]自动验证拒绝john，400响应指出期望bool、实际str。](https://www.unfinished-era.com/wp-content/uploads/2026/10/wwj-2496-6.png)
让参数可选
只声明布尔类型却没有默认值时，done仍是必填参数，省略它会报错。要让缺省请求返回整个列表，应声明 done: FromQuery[Optional[bool]] = None，或3.10+写法FromQuery[bool | None] = None；在done is None时返回所有项，否则过滤。若想默认只看未完成任务，也可以把默认值设成False。可选类型与默认值是共同的配置，不应只加Optional就假定参数不再必填。
交互式API文档
运行应用后，/schema/swagger 展示自动生成的路由与模型，并可发送和查看请求。Swagger利用Litestar从处理器生成的OpenAPI schema确定字段、参数与响应。原图演示done=true请求；同一schema也可在 /schema/redoc 与 /schema/elements 通过ReDoc、Stoplight Elements查看。访问页面本身不代表已经测试所有路径。
三、接收请求体与动态路径
新增待办使用 @post("/")。名为data的参数用于接收请求体；注解为dict时，Litestar按JSON读取并按类型转换，原例把data追加到TODO_LIST并返回整个列表。浏览器直接访问URL通常是GET，不能据此调用POST；可用Swagger发送示例JSON并查看201响应。
将data注解改成TodoItem数据类后，框架会将JSON字段构造成TodoItem，Swagger也能显示title、done而不是空字典结构；缺少title会得到有用的验证错误。数据类示例改善结构和验证，但生产系统仍需要长度、取值与业务约束。
更新特定待办用路径参数。例如 @get("/{name:str}") 配合 name: FromPath[str] 可接收/john的字符串john。花括号中的名称与参数名称要对应，冒号后的类型控制转换；str接收字符串，int尝试整数。它是一段动态匹配槽，而不是固定路径。
PUT处理器使用 /{item_title:str}，接收路径标题与TodoItem请求体。get_todo_by_title遍历内存列表，找不到就抛NotFoundException（404）；找到后修改title与done并返回列表。这里标题既是可变数据又是定位键，重复标题和改名会使身份不稳定。原教程没有删除端点，也没有用户级访问控制。
四、组装内存应用
第四章把数据类、三条初始记录、按标题查找、可选done过滤、POST追加、PUT更新与app注册合在一份独立app.py。GET返回全部或筛选后的列表；POST通过data解析JSON为TodoItem；PUT由FromPath接收item_title并应用新值。三个处理器一起注册给Litestar，再由ASGI服务器启动。
这份应用展示框架基础，不具备持久化：进程结束数据消失，多worker的模块级列表也各自独立。附录保留其三套Python注解版本。进一步的框架特性可沿原文Usage Guide学习，下一部分专门解决数据库持久化。
五、引入SQLAlchemy：先显式管理全部资源
第五章先展示一份较长但行为接近前例的应用：使用DeclarativeBase和Mapped定义todo_items表，title为主键，done为布尔列；通过create_async_engine连接相对路径SQLite数据库 sqlite+aiosqlite:///todo.sqlite。它引入应用state和lifespan，后面再逐步用依赖注入和插件简化。
引擎生命周期与建表
db_connection是asynccontextmanager：先从app.state读取engine，没有时创建并保存；在engine.begin上下文中通过conn.run_sync调用Base.metadata.create_all；yield让应用运行，finally释放engine。create_all创建缺失的表，已按模型建立的表不会重复创建，但这不是数据库结构迁移工具。
app.state在生命周期函数中保存引擎，处理器则通过注解为State的state参数访问它。async_sessionmaker设置expire_on_commit=False，处理器按state.engine建立会话。源文有echo=True用于输出SQL，真实环境应审查日志中是否包含敏感数据。
查询、事务与手工序列化
get_todo_by_title使用select和where查询标题，再以scalar_one取得对象；NoResultFound映射成404。get_todo_list可按done条件筛选并返回scalars。原例通过TodoType、TodoCollectionType类型别名表示JSON边界，serialize_todo显式返回title与done字典，因为还没有安装自动处理ORM对象的插件。
POST在session.begin事务中加入新对象，IntegrityError映射为409并指出标题已存在。PUT读取实体后修改字段。与内存版本不同，POST和PUT现在返回新增或更新的单条对象，不再返回整个列表。原例每个处理器都重复创建会话，这正是下一章要消除的重复。
六、用依赖注入提供事务
provide_transaction异步生成器创建会话，进入session.begin，yield给处理器；完整性错误集中映射为409。把它以transaction键注册到app.dependencies，处理器使用同名参数，并在此阶段以NamedDependency[AsyncSession]标记，让Litestar按依赖键注入该值。
GET不再自己创建会话，直接查询并序列化；POST构造TodoItem并加入传入会话；PUT查找和修改。会话的创建、事务结束及IntegrityError处理由provider负责，错误处理从仅覆盖新增扩大到所有使用该会话的操作。源文的detail=str(exc)可能向客户端泄露SQL、表名和输入，部署时应返回固定的通用错误，细节仅写受控日志。
这种简化没有实现外部副作用的原子性。如果请求还发送邮件或调用其他服务，不能依赖数据库事务自动撤销这些动作。取消、提交失败、异常响应和连接关闭均需要在选定版本中测试；本文没有运行这些测试。
七、用序列化插件直接收发ORM对象
加入SQLAlchemySerializationPlugin后，处理器可以接收、返回TodoItem模型，不再需要TodoType、TodoCollectionType和serialize_todo。GET直接返回ORM对象列表；POST把已构造的模型data加入事务并返回data；PUT读取现有对象，使用data.title与data.done更新并返回该对象。
这一阶段仍保留手动db_connection和provide_transaction。原页完整代码使用普通AsyncSession注解，而前一DI页使用NamedDependency；这是源页版本/阶段差异，不能把两种写法任意混合。附录保留各阶段原例，具体依赖标记行为必须按锁定组合核验。
八、用初始化插件管理引擎与会话
SQLAlchemyInitPlugin进一步接管两层生命周期：引擎与应用同生共灭，会话与请求同生共灭。SQLAlchemyAsyncConfig提供连接字符串、Base.metadata和create_all=True，使插件创建需要的表，并提供db_session依赖。
provide_transaction接收插件提供的db_session，再管理session.begin与错误映射，因此不再自行创建会话；app不再声明手动db_connection生命周期。此阶段仍同时列出SQLAlchemySerializationPlugin和SQLAlchemyInitPlugin，前者负责ORM序列化，后者负责资源初始化。
九、最终组合与回顾
SQLAlchemyPlugin组合上述两插件，让配置更短。最后一页的SQLAlchemyAsyncConfig保留SQLite、metadata、create_all，并设置before_send_handler="autocommit"；应用仍注册transaction provider和GET/POST/PUT三路由。附录中的最终完整程序原样保留，包含404/409映射、可选done过滤、标题主键和自动建表。
总结数据流：TodoItem描述存储模型；db_session由插件注入provider；transaction提供处理器所用事务；查询辅助函数只构造SQLAlchemy表达式，不拼接SQL文本；处理器是HTTP边界；插件管理引擎、会话与ORM的JSON转换。最终例的普通AsyncSession标记及autocommit与事务上下文配合，需要按准确版本验证提交/回滚时机，不能仅凭文档片段断言运行通过。
生产改造至少还要处理不可变ID、标题长度与空值、身份认证、对象级与租户级权限、并发冲突、迁移、持久卷、备份恢复和SQLite写并发。完整性异常不能直接回传数据库细节；create_all不是迁移或备份。此系列没有删除接口，也没有证明多副本可通过共享SQLite文件安全扩展。所有SQL和HTTP示例仅静态核查。
九章原始代码对照
下面按原页顺序保留代码，包括独立阶段的完整应用、用于讲解的局部高亮片段及Python语法变体。局部片段并不是可单独运行的程序；同名定义不能累计拼接。原例的文件数据库、echo日志、原始异常文案等保留用于核对，上述安全注释另列。未执行任何代码。
原第1页：0-application-basics
原代码 1-1
from litestar import Litestar, get
@get("/")
async def hello_world() -> str:
return "Hello, world!"
app = Litestar([hello_world])
原代码 1-2
from litestar import Litestar, get
@get("/")
async def hello_world() -> str:
return "Hello, world!"
app = Litestar([hello_world])
原代码 1-3
async def hello_world() -> str:
return "Hello, world!"
hello_world = get("/")(hello_world)
原代码 1-4
from litestar import Litestar, get
@get("/")
async def hello_world() -> str:
return "Hello, world!"
app = Litestar([hello_world])
原代码 1-5
from litestar import Litestar, get
@get("/")
async def hello_world() -> str:
return "Hello, world!"
app = Litestar([hello_world])
原第2页：1-accessing-the-list
原代码 2-1
from typing import Dict, List, Union
from litestar import Litestar, get
TODO_LIST: List[Dict[str, Union[str, bool]]] = [
{"title": "Start writing TODO list", "done": True},
{"title": "???", "done": False},
{"title": "Profit", "done": False},
]
@get("/")
async def get_list() -> List[Dict[str, Union[str, bool]]]:
return TODO_LIST
app = Litestar([get_list])
原代码 2-2
from typing import Union
from litestar import Litestar, get
TODO_LIST: list[dict[str, Union[str, bool]]] = [
{"title": "Start writing TODO list", "done": True},
{"title": "???", "done": False},
{"title": "Profit", "done": False},
]
@get("/")
async def get_list() -> list[dict[str, Union[str, bool]]]:
return TODO_LIST
app = Litestar([get_list])
原代码 2-3
from litestar import Litestar, get
TODO_LIST: list[dict[str, str | bool]] = [
{"title": "Start writing TODO list", "done": True},
{"title": "???", "done": False},
{"title": "Profit", "done": False},
]
@get("/")
async def get_list() -> list[dict[str, str | bool]]:
return TODO_LIST
app = Litestar([get_list])
原代码 2-4
async def get_list() -> List[Dict[str, Union[str, bool]]]:
原代码 2-5
from dataclasses import dataclass
from typing import List
from litestar import Litestar, get
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list() -> List[TodoItem]:
return TODO_LIST
app = Litestar([get_list])
原代码 2-6
from dataclasses import dataclass
from litestar import Litestar, get
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list() -> list[TodoItem]:
return TODO_LIST
app = Litestar([get_list])
原代码 2-7
from dataclasses import dataclass
from typing import List
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[str]) -> List[TodoItem]:
if done == "1":
return [item for item in TODO_LIST if item.done]
return [item for item in TODO_LIST if not item.done]
app = Litestar([get_list])
原代码 2-8
from dataclasses import dataclass
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[str]) -> list[TodoItem]:
if done == "1":
return [item for item in TODO_LIST if item.done]
return [item for item in TODO_LIST if not item.done]
app = Litestar([get_list])
原代码 2-9
from dataclasses import dataclass
from typing import List
from litestar import Litestar, get
from litestar.exceptions import HTTPException
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[str]) -> List[TodoItem]:
if done == "1":
return [item for item in TODO_LIST if item.done]
if done == "0":
return [item for item in TODO_LIST if not item.done]
raise HTTPException(f"Invalid query parameter value: {done!r}", status_code=400)
app = Litestar([get_list])
原代码 2-10
from dataclasses import dataclass
from litestar import Litestar, get
from litestar.exceptions import HTTPException
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[str]) -> list[TodoItem]:
if done == "1":
return [item for item in TODO_LIST if item.done]
if done == "0":
return [item for item in TODO_LIST if not item.done]
raise HTTPException(f"Invalid query parameter value: {done!r}", status_code=400)
app = Litestar([get_list])
原代码 2-11
if done == "1":
return [item for item in TODO_LIST if item.done]
原代码 2-12
if done == "0":
return [item for item in TODO_LIST if not item.done]
原代码 2-13
raise HTTPException(f"Invalid query parameter value: {done!r}", status_code=400)
原代码 2-14
from dataclasses import dataclass
from typing import List
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[bool]) -> List[TodoItem]:
return [item for item in TODO_LIST if item.done == done]
app = Litestar([get_list])
原代码 2-15
from dataclasses import dataclass
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[bool]) -> list[TodoItem]:
return [item for item in TODO_LIST if item.done == done]
app = Litestar([get_list])
原代码 2-16
async def get_list(done: FromQuery[bool]) -> List[TodoItem]:
原代码 2-17
return [item for item in TODO_LIST if item.done == done]
原代码 2-18
from dataclasses import dataclass
from typing import List, Optional
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[Optional[bool]] = None) -> List[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
app = Litestar([get_list])
原代码 2-19
from dataclasses import dataclass
from typing import Optional
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[Optional[bool]] = None) -> list[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
app = Litestar([get_list])
原代码 2-20
from dataclasses import dataclass
from litestar import Litestar, get
from litestar.params import FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
@get("/")
async def get_list(done: FromQuery[bool | None] = None) -> list[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
app = Litestar([get_list])
原第3页：2-interacting-with-the-list
原代码 3-1
from typing import Any, Dict, List, Union
from litestar import Litestar, post
TODO_LIST: List[Dict[str, Union[str, bool]]] = []
@post("/")
async def add_item(data: Dict[str, Any]) -> List[Dict[str, Union[str, bool]]]:
TODO_LIST.append(data)
return TODO_LIST
app = Litestar([add_item])
原代码 3-2
from typing import Any, Union
from litestar import Litestar, post
TODO_LIST: list[dict[str, Union[str, bool]]] = []
@post("/")
async def add_item(data: dict[str, Any]) -> list[dict[str, Union[str, bool]]]:
TODO_LIST.append(data)
return TODO_LIST
app = Litestar([add_item])
原代码 3-3
from typing import Any
from litestar import Litestar, post
TODO_LIST: list[dict[str, str | bool]] = []
@post("/")
async def add_item(data: dict[str, Any]) -> list[dict[str, str | bool]]:
TODO_LIST.append(data)
return TODO_LIST
app = Litestar([add_item])
原代码 3-4
from dataclasses import dataclass
from typing import List
from litestar import Litestar, post
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = []
@post("/")
async def add_item(data: TodoItem) -> List[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
app = Litestar([add_item])
原代码 3-5
from dataclasses import dataclass
from litestar import Litestar, post
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = []
@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
app = Litestar([add_item])
原代码 3-6
@get("/{name:str}")
async def greeter(name: FromPath[str]) -> str:
return "Hello, " + name
原代码 3-7
from dataclasses import dataclass
from typing import List
from litestar import Litestar, put
from litestar.exceptions import NotFoundException
from litestar.params import FromPath
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
def get_todo_by_title(todo_name: str) -> TodoItem:
for item in TODO_LIST:
if item.title == todo_name:
return item
raise NotFoundException(detail=f"TODO {todo_name!r} not found")
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> List[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
app = Litestar([update_item])
原代码 3-8
from dataclasses import dataclass
from litestar import Litestar, put
from litestar.exceptions import NotFoundException
from litestar.params import FromPath
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
def get_todo_by_title(todo_name: str) -> TodoItem:
for item in TODO_LIST:
if item.title == todo_name:
return item
raise NotFoundException(detail=f"TODO {todo_name!r} not found")
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> list[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
app = Litestar([update_item])
原第4页：3-assembling-the-app
原代码 4-1
from dataclasses import dataclass
from typing import List, Optional
from litestar import Litestar, get, post, put
from litestar.exceptions import NotFoundException
from litestar.params import FromPath, FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: List[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
def get_todo_by_title(todo_name: str) -> TodoItem:
for item in TODO_LIST:
if item.title == todo_name:
return item
raise NotFoundException(detail=f"TODO {todo_name!r} not found")
@get("/")
async def get_list(done: FromQuery[Optional[bool]] = None) -> List[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
@post("/")
async def add_item(data: TodoItem) -> List[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> List[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
app = Litestar([get_list, add_item, update_item])
原代码 4-2
from dataclasses import dataclass
from typing import Optional
from litestar import Litestar, get, post, put
from litestar.exceptions import NotFoundException
from litestar.params import FromPath, FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
def get_todo_by_title(todo_name: str) -> TodoItem:
for item in TODO_LIST:
if item.title == todo_name:
return item
raise NotFoundException(detail=f"TODO {todo_name!r} not found")
@get("/")
async def get_list(done: FromQuery[Optional[bool]] = None) -> list[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> list[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
app = Litestar([get_list, add_item, update_item])
原代码 4-3
from dataclasses import dataclass
from litestar import Litestar, get, post, put
from litestar.exceptions import NotFoundException
from litestar.params import FromPath, FromQuery
@dataclass
class TodoItem:
title: str
done: bool
TODO_LIST: list[TodoItem] = [
TodoItem(title="Start writing TODO list", done=True),
TodoItem(title="???", done=False),
TodoItem(title="Profit", done=False),
]
def get_todo_by_title(todo_name: str) -> TodoItem:
for item in TODO_LIST:
if item.title == todo_name:
return item
raise NotFoundException(detail=f"TODO {todo_name!r} not found")
@get("/")
async def get_list(done: FromQuery[bool | None] = None) -> list[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
@post("/")
async def add_item(data: TodoItem) -> list[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> list[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
app = Litestar([get_list, add_item, update_item])
原代码 4-4
@get("/")
async def get_list(done: FromQuery[Optional[bool]] = None) -> List[TodoItem]:
if done is None:
return TODO_LIST
return [item for item in TODO_LIST if item.done == done]
原代码 4-5
@post("/")
async def add_item(data: TodoItem) -> List[TodoItem]:
TODO_LIST.append(data)
return TODO_LIST
原代码 4-6
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoItem) -> List[TodoItem]:
todo_item = get_todo_by_title(item_title)
todo_item.title = data.title
todo_item.done = data.done
return TODO_LIST
原代码 4-7
app = Litestar([get_list, add_item, update_item])
原第5页：0-introduction
原代码 5-1
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, Dict, List, Optional, Sequence
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = Dict[str, Any]
TodoCollectionType = List[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[Optional[bool]], session: AsyncSession) -> Sequence[TodoItem]:
query = select(TodoItem)
if done is not None:
query = query.where(TodoItem.done.is_(done))
result = await session.execute(query)
return result.scalars().all()
@get("/")
async def get_list(state: State, done: FromQuery[Optional[bool]] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
return [serialize_todo(todo) for todo in await get_todo_list(done, session)]
@post("/")
async def add_item(data: TodoType, state: State) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
session.add(new_todo)
except IntegrityError as e:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=f"TODO {new_todo.title!r} already exists",
) from e
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar([get_list, add_item, update_item], lifespan=[db_connection])
原代码 5-2
from contextlib import asynccontextmanager
from typing import Any, Optional
from collections.abc import AsyncGenerator, Sequence
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = dict[str, Any]
TodoCollectionType = list[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[Optional[bool]], session: AsyncSession) -> Sequence[TodoItem]:
query = select(TodoItem)
if done is not None:
query = query.where(TodoItem.done.is_(done))
result = await session.execute(query)
return result.scalars().all()
@get("/")
async def get_list(state: State, done: FromQuery[Optional[bool]] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
return [serialize_todo(todo) for todo in await get_todo_list(done, session)]
@post("/")
async def add_item(data: TodoType, state: State) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
session.add(new_todo)
except IntegrityError as e:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=f"TODO {new_todo.title!r} already exists",
) from e
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar([get_list, add_item, update_item], lifespan=[db_connection])
原代码 5-3
from contextlib import asynccontextmanager
from typing import Any
from collections.abc import AsyncGenerator, Sequence
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = dict[str, Any]
TodoCollectionType = list[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[bool | None], session: AsyncSession) -> Sequence[TodoItem]:
query = select(TodoItem)
if done is not None:
query = query.where(TodoItem.done.is_(done))
result = await session.execute(query)
return result.scalars().all()
@get("/")
async def get_list(state: State, done: FromQuery[bool | None] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
return [serialize_todo(todo) for todo in await get_todo_list(done, session)]
@post("/")
async def add_item(data: TodoType, state: State) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
session.add(new_todo)
except IntegrityError as e:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=f"TODO {new_todo.title!r} already exists",
) from e
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar([get_list, add_item, update_item], lifespan=[db_connection])
原代码 5-4
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
原代码 5-5
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
原代码 5-6
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
原代码 5-7
@get("/")
async def get_list(state: State, done: FromQuery[Optional[bool]] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
原代码 5-8
from typing import Any, Dict
TodoType = Dict[str, Any]
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
原代码 5-9
from typing import Any
TodoType = dict[str, Any]
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
原第6页：1-provide-session-with-di
原代码 6-1
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, Dict, List, Optional
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.di import NamedDependency
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = Dict[str, Any]
TodoCollectionType = List[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite")
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield session
except IntegrityError as exc:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=str(exc),
) from exc
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[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 result.scalars().all()
@get("/")
async def get_list(
transaction: NamedDependency[AsyncSession], done: FromQuery[Optional[bool]] = None
) -> TodoCollectionType:
return [serialize_todo(todo) for todo in await get_todo_list(done, transaction)]
@post("/")
async def add_item(data: TodoType, transaction: NamedDependency[AsyncSession]) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
transaction.add(new_todo)
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(
item_title: FromPath[str], data: TodoType, transaction: NamedDependency[AsyncSession]
) -> TodoType:
todo_item = await get_todo_by_title(item_title, transaction)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
)
原代码 6-2
from contextlib import asynccontextmanager
from typing import Any, Optional
from collections.abc import AsyncGenerator
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.di import NamedDependency
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = dict[str, Any]
TodoCollectionType = list[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite")
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield session
except IntegrityError as exc:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=str(exc),
) from exc
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[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 result.scalars().all()
@get("/")
async def get_list(
transaction: NamedDependency[AsyncSession], done: FromQuery[Optional[bool]] = None
) -> TodoCollectionType:
return [serialize_todo(todo) for todo in await get_todo_list(done, transaction)]
@post("/")
async def add_item(data: TodoType, transaction: NamedDependency[AsyncSession]) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
transaction.add(new_todo)
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(
item_title: FromPath[str], data: TodoType, transaction: NamedDependency[AsyncSession]
) -> TodoType:
todo_item = await get_todo_by_title(item_title, transaction)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
)
原代码 6-3
from contextlib import asynccontextmanager
from typing import Any
from collections.abc import AsyncGenerator
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.di import NamedDependency
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.status_codes import HTTP_409_CONFLICT
TodoType = dict[str, Any]
TodoCollectionType = list[TodoType]
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
done: Mapped[bool]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite")
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield session
except IntegrityError as exc:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=str(exc),
) from exc
def serialize_todo(todo: TodoItem) -> TodoType:
return {"title": todo.title, "done": todo.done}
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[bool | None], 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 result.scalars().all()
@get("/")
async def get_list(
transaction: NamedDependency[AsyncSession], done: FromQuery[bool | None] = None
) -> TodoCollectionType:
return [serialize_todo(todo) for todo in await get_todo_list(done, transaction)]
@post("/")
async def add_item(data: TodoType, transaction: NamedDependency[AsyncSession]) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
transaction.add(new_todo)
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(
item_title: FromPath[str], data: TodoType, transaction: NamedDependency[AsyncSession]
) -> TodoType:
todo_item = await get_todo_by_title(item_title, transaction)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
)
原代码 6-4
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield session
except IntegrityError as exc:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=str(exc),
) from exc
原代码 6-5
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
原代码 6-6
@get("/")
async def get_list(
transaction: NamedDependency[AsyncSession], done: FromQuery[Optional[bool]] = None
原代码 6-7
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield session
except IntegrityError as exc:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=str(exc),
) from exc
原代码 6-8
@get("/")
async def get_list(
transaction: NamedDependency[AsyncSession], done: FromQuery[Optional[bool]] = None
) -> TodoCollectionType:
return [serialize_todo(todo) for todo in await get_todo_list(done, transaction)]
@post("/")
async def add_item(data: TodoType, transaction: NamedDependency[AsyncSession]) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
transaction.add(new_todo)
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(
item_title: FromPath[str], data: TodoType, transaction: NamedDependency[AsyncSession]
) -> TodoType:
todo_item = await get_todo_by_title(item_title, transaction)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
原代码 6-9
@get("/")
async def get_list(state: State, done: FromQuery[Optional[bool]] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
return [serialize_todo(todo) for todo in await get_todo_list(done, session)]
@post("/")
async def add_item(data: TodoType, state: State) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
session.add(new_todo)
except IntegrityError as e:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=f"TODO {new_todo.title!r} already exists",
) from e
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
原第7页：2-serialization-plugin
原代码 7-1
from contextlib import asynccontextmanager
from typing import AsyncGenerator, List, Optional
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin
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]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield 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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
plugins=[SQLAlchemySerializationPlugin()],
)
原代码 7-2
from contextlib import asynccontextmanager
from typing import Optional
from collections.abc import AsyncGenerator
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin
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]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield 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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
plugins=[SQLAlchemySerializationPlugin()],
)
原代码 7-3
from contextlib import asynccontextmanager
from collections.abc import AsyncGenerator
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin
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]
@asynccontextmanager
async def db_connection(app: Litestar) -> AsyncGenerator[None, None]:
engine = getattr(app.state, "engine", None)
if engine is None:
engine = create_async_engine("sqlite+aiosqlite:///todo.sqlite", echo=True)
app.state.engine = engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
try:
yield
finally:
await engine.dispose()
sessionmaker = async_sessionmaker(expire_on_commit=False)
async def provide_transaction(state: State) -> AsyncGenerator[AsyncSession, None]:
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
yield 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: FromPath[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: FromQuery[bool | None], 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: FromQuery[bool | None] = 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: FromPath[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
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
plugins=[SQLAlchemySerializationPlugin()],
)
原代码 7-4
from contextlib import asynccontextmanager
from typing import AsyncGenerator, List, Optional
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemySerializationPlugin
return list(result.scalars().all())
@get("/")
async def get_list(transaction: AsyncSession, done: FromQuery[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: FromPath[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
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
lifespan=[db_connection],
plugins=[SQLAlchemySerializationPlugin()],
原代码 7-5
from contextlib import asynccontextmanager
from typing import Any, AsyncGenerator, Dict, List, Optional, Sequence
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError, NoResultFound
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from litestar import Litestar, get, post, put
from litestar.datastructures import State
from litestar.exceptions import ClientException, NotFoundException
from litestar.params import FromPath, FromQuery
return result.scalars().all()
@get("/")
async def get_list(state: State, done: FromQuery[Optional[bool]] = None) -> TodoCollectionType:
async with sessionmaker(bind=state.engine) as session:
return [serialize_todo(todo) for todo in await get_todo_list(done, session)]
@post("/")
async def add_item(data: TodoType, state: State) -> TodoType:
new_todo = TodoItem(title=data["title"], done=data["done"])
async with sessionmaker(bind=state.engine) as session:
try:
async with session.begin():
session.add(new_todo)
except IntegrityError as e:
raise ClientException(
status_code=HTTP_409_CONFLICT,
detail=f"TODO {new_todo.title!r} already exists",
) from e
return serialize_todo(new_todo)
@put("/{item_title:str}")
async def update_item(item_title: FromPath[str], data: TodoType, state: State) -> TodoType:
async with sessionmaker(bind=state.engine) as session, session.begin():
todo_item = await get_todo_by_title(item_title, session)
todo_item.title = data["title"]
todo_item.done = data["done"]
return serialize_todo(todo_item)
原第8页：3-init-plugin
原代码 8-1
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import (
SQLAlchemyAsyncConfig,
SQLAlchemyInitPlugin,
SQLAlchemySerializationPlugin,
)
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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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
)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
plugins=[
SQLAlchemySerializationPlugin(),
SQLAlchemyInitPlugin(db_config),
],
)
原代码 8-2
from typing import Optional
from collections.abc import AsyncGenerator
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import (
SQLAlchemyAsyncConfig,
SQLAlchemyInitPlugin,
SQLAlchemySerializationPlugin,
)
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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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
)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
plugins=[
SQLAlchemySerializationPlugin(),
SQLAlchemyInitPlugin(db_config),
],
)
原代码 8-3
from collections.abc import AsyncGenerator
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import (
SQLAlchemyAsyncConfig,
SQLAlchemyInitPlugin,
SQLAlchemySerializationPlugin,
)
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: FromPath[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: FromQuery[bool | None], 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: FromQuery[bool | None] = 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: FromPath[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
)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
plugins=[
SQLAlchemySerializationPlugin(),
SQLAlchemyInitPlugin(db_config),
],
)
原第9页：4-final-touches-and-recap
原代码 9-1
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.params import FromPath, FromQuery
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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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)],
)
原代码 9-2
from typing import Optional
from collections.abc import AsyncGenerator
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.params import FromPath, FromQuery
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: FromPath[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: FromQuery[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: FromQuery[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: FromPath[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)],
)
原代码 9-3
from collections.abc import AsyncGenerator
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.params import FromPath, FromQuery
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: FromPath[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: FromQuery[bool | None], 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: FromQuery[bool | None] = 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: FromPath[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)],
)
原代码 9-4
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
class Base(DeclarativeBase): ...
class TodoItem(Base):
__tablename__ = "todo_items"
title: Mapped[str] = mapped_column(primary_key=True)
原代码 9-5
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
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),
原代码 9-6
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
) from exc
async def get_todo_by_title(todo_name: FromPath[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: FromQuery[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)
原代码 9-7
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
return list(result.scalars().all())
@get("/")
async def get_list(transaction: AsyncSession, done: FromQuery[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: FromPath[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
原代码 9-8
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.params import FromPath, FromQuery
from litestar.plugins.sqlalchemy import SQLAlchemyAsyncConfig, SQLAlchemyPlugin
)
app = Litestar(
[get_list, add_item, update_item],
dependencies={"transaction": provide_transaction},
plugins=[SQLAlchemyPlugin(db_config)],
来源、修改与许可证
上述九页由Litestar-Org维护，页面版权为©2025 Litestar-Org。中文翻译、结构合并和静态技术注释：未完纪，2026-10-05；原代码及12张原图按其源版本保留，没有把源截图冒充新测试。公开保留官方MIT版权与许可全文，原网页版权年份与当前仓库版权范围分别说明，不相互替换。
MIT License 全文
The MIT License (MIT)
Copyright (c) 2021, 2022, 2023, 2024, 2025, 2026 Litestar Org.
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.© 版权声明
文章版权归作者所有，未经允许请勿转载。
THE END
暂无评论内容