使用 Litestar 的 SQLAlchemySerializationPlugin 直接在处理器中收发 SQLAlchemy 模型【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇技术指南聚焦于 Litestar 项目中集成 SQLAlchemy 的核心进阶话题借助advanced_alchemy.extensions.litestar.plugins.SQLAlchemySerializationPlugin下文简称“序列化插件”让路由处理器直接接收与返回 SQLAlchemy ORM 模型从而移除手写的类型别名和serialize_todo()转换函数。它是 Litestar 官方 SQLAlchemy 教程docs/tutorials/sqlalchemy中的第二步承接前一篇使用依赖注入提供数据库会话的改造。读完本文你将掌握序列化插件的安装、注册方式、改造前后处理器代码的对比以及该插件与SQLAlchemyInitPlugin、SQLAlchemyPlugin的关系能直接在自己的 Litestar SQLAlchemy 应用中落地“模型直进直出”的写法。为什么需要序列化插件回顾改造前的痛点在未使用任何 SQLAlchemy 插件时完整代码见 full_app_no_plugins.py应用虽然已经能运行但存在一个明显的别扭之处Litestar 无法自动对 SQLAlchemy 模型做 (反)序列化。原文档在“Introduction”一节docs/tutorials/sqlalchemy/0-introduction.rst明确指出由于改用 SQLAlchemy 模型承载数据必须将模型转换成 Litestar 能够序列化的类型为此代码里引入了一堆“脚手架”from typing import Any TodoType dict[str, Any] TodoCollectionType list[TodoType]以及一个负责把 ORM 对象转成 dict 的辅助函数def serialize_todo(todo: TodoItem) - TodoType: return {title: todo.title, done: todo.done}于是每个处理器都不得不在“模型”与“可序列化类型”之间反复搬运接收时用data[title]、data[done]手工拆包构造TodoItem返回时再用serialize_todo(new_todo)手工打包成 dict例如post(/) async def add_item(data: TodoType, state: State) - TodoType: new_todo TodoItem(titledata[title], donedata[done]) async with sessionmaker(bindstate.engine) as session: ... return serialize_todo(new_todo)而此前教程docs/tutorials/sqlalchemy/1-provide-session-with-di.rst也提示过最初的 TODO 应用使用 Python dataclass 建模Litestar 原生支持 dataclass 的 (反)序列化换成 SQLAlchemy 模型后这个便利就丢了。序列化插件正是要把这个能力“找回来”。引入序列化插件两行改动带来模型直进直出改造后的完整应用代码位于 full_app_with_serialization_plugin.py其核心改动可以浓缩为两点导入插件from advanced_alchemy.extensions.litestar import SQLAlchemySerializationPlugin注册到应用app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, lifespan[db_connection], plugins[SQLAlchemySerializationPlugin()], )仅此而已。注册之后处理器签名可以直接使用TodoItem模型作为参数类型与返回类型Litestar 会在请求进入时把 JSON 请求体反序列化为TodoItem实例在响应离开时把TodoItem以及list[TodoItem]序列化为 JSON。同时由于模型已经可以直接进出处理器代码中还顺手删掉了三样东西TodoType别名TodoCollectionType别名serialize_todo()辅助函数。改造前需要from typing import Any和from collections.abc import Sequence改造后这两个导入也不再需要整个实现更加简洁。改造前后的处理器对比原文档专门用“Compare handlers before and after Serialization Plugin”一节做了并排对比After 取 full_app_with_serialization_plugin.py 与 同文件第 73-99 行Before 取 full_app_no_plugins.py 与 同文件第 67-100 行。下表汇总两版处理器在“接收、内部处理、返回”三个环节的差异环节Before无插件After序列化插件接收请求体data: TodoType手动TodoItem(titledata[title], donedata[done])拆包data: TodoItem直接使用模型实例操作数据在处理器内手动开 session、手动管理事务与异常依赖注入的transaction: AsyncSession事务与IntegrityError处理集中在provide_transaction()返回响应serialize_todo(todo)手工转 dict返回TodoType/TodoCollectionType直接返回TodoItem/list[TodoItem]改造后的三个处理器完整代码如下get(/) async def get_list(transaction: AsyncSession, done: 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: 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可以看到add_item()从“拆包 构造 手工序列化”的 6 行逻辑缩成了“transaction.add(data)后直接返回 data”的 2 行update_item()也直接用data.title、data.done属性赋值不再有data[title]式的字典索引。这正是原文档强调的“Now we can receive and return our SQLAlchemy data models directly to and from our handler”。值得一提的细节改造前add_item()与update_item()路由返回的是“被添加/被更新的单条数据”而非整个集合这一行为约定在改造后原样保留——现在返回类型直接声明为TodoItem语义上反而更清晰行为细节讨论见 docs/tutorials/sqlalchemy/0-introduction.rst。插件的最小用法与安装方式序列化插件本身不需要任何配置参数即可工作。仓库中另有一个精简示例 sqlalchemy_async_serialization_plugin.py展示了它的最小用法——一个POST /处理器直接接收TodoItem并返回list[TodoItem]from __future__ import annotations from advanced_alchemy.extensions.litestar import SQLAlchemySerializationPlugin from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from litestar import Litestar, post class Base(DeclarativeBase): ... class TodoItem(Base): __tablename__ todo_item title: Mapped[str] mapped_column(primary_keyTrue) done: Mapped[bool] post(/) async def add_item(data: TodoItem) - list[TodoItem]: return [data] app Litestar(route_handlers[add_item], plugins[SQLAlchemySerializationPlugin()])注意该示例没有配置任何数据库引擎或会话——这说明序列化插件只负责“模型 - JSON”的类型映射与 (反)序列化不负责数据库连接的创建与管理数据库相关工具引擎、会话、事务依赖等由SQLAlchemyInitPlugin提供两者职责正交。关于这一点docs 的插件索引docs/usage/databases/sqlalchemy/plugins/index.rst将其明确划分为SQLAlchemyPlugin完整的 SQLAlchemy 支持Init Serialization 的组合SQLAlchemyInitPlugin应用级工具引擎/会话管理SQLAlchemySerializationPlugin序列化支持。安装方面docs 在 docs/usage/databases/sqlalchemy/plugins/sqlalchemy_plugin.rst 给出了推荐方式运行pip install litestar[sqlalchemy]即可把 SQLAlchemy 及其相关插件一并装上同时该文档特别注明插件仅兼容 SQLAlchemy 2.0。本教程示例使用的驱动是 SQLite 异步驱动sqliteaiosqlite涉及advanced_alchemy库其文档由Advanced Alchemy项目维护。前置条件回顾会话依赖与事务处理序列化插件解决的是“数据形态”问题但应用里仍然保留着上一节教程建立的依赖注入机制docs/tutorials/sqlalchemy/1-provide-session-with-di.rst两者共同工作sessionmaker async_sessionmaker(expire_on_commitFalse) async def provide_transaction(state: State) - AsyncGenerator[AsyncSession, None]: async with sessionmaker(bindstate.engine) as session: try: async with session.begin(): yield session except IntegrityError as exc: raise ClientException( status_codeHTTP_409_CONFLICT, detailstr(exc), ) from exc该依赖通过dependencies{transaction: provide_transaction}注册处理器里参数名为transaction即自动注入AsyncSession在早期版本中还需配合NamedDependency标记本教程示例直接按参数名匹配。db_connection()生命周期上下文管理器则负责在应用启动时创建异步引擎、执行Base.metadata.create_all建表并在关闭时engine.dispose()释放资源。也就是说序列化插件与依赖注入各司其职DI 解决“session 从哪来、事务怎么管、异常怎么统一转 HTTP 409”序列化插件解决“模型怎么进、怎么出”。两者叠加后处理器代码只剩下纯粹的业务逻辑。下一步用 SQLAlchemyInitPlugin 进一步消除脚手架原文档在“Next steps”一节提示目前应用中还残留着db_connection()生命周期上下文管理器与provide_transaction()依赖提供者这类“脚手架”代码下一步将引入SQLAlchemyInitPluginadvanced_alchemy.extensions.litestar.plugins.SQLAlchemyInitPlugin来接管引擎与会话的配置管理。这正是教程的第三节docs/tutorials/sqlalchemy/3-init-plugin.rst的内容。此外值得提前了解的是SQLAlchemySerializationPlugin与SQLAlchemyInitPlugin经常成对出现以至于 Advanced Alchemy 提供了二者的合并版SQLAlchemyPlugin——正如教程收尾篇docs/tutorials/sqlalchemy/4-final-touches-and-recap.rst所说它“is a combination of the two”可以直接替代两个插件分别注册让Litestar(..., plugins[SQLAlchemyPlugin()])一行配置搞定全部 SQLAlchemy 集成。完整的合并用法参见 docs/usage/databases/sqlalchemy/plugins/sqlalchemy_plugin.rst其中提供了 Async 与 Sync 两套完整示例以及litestar run启动后curl -X POST提交{title: ..., done: false}的验证方式。小结本文对应官方教程 docs/tutorials/sqlalchemy/2-serialization-plugin.rst 的核心内容你可以通过以下要点快速回顾SQLAlchemySerializationPlugin来自advanced_alchemy.extensions.litestar注册方式为plugins[SQLAlchemySerializationPlugin()]注册后处理器可直接以 SQLAlchemy 模型作为请求体参数类型与响应返回类型无需手写类型别名或serialize_todo()之类的转换函数该插件只负责 (反)序列化不管理数据库连接引擎/会话/事务由依赖注入与生命周期钩子或SQLAlchemyInitPlugin负责插件兼容 SQLAlchemy 2.0推荐通过pip install litestar[sqlalchemy]安装若同时使用 Init 与 Serialization 两个插件可进一步简化为单个SQLAlchemyPlugin。相关文件索引改造后完整示例 full_app_with_serialization_plugin.py、改造前示例 full_app_no_plugins.py、插件总览 docs/usage/databases/sqlalchemy/plugins/index.rst、插件完整用法 docs/usage/databases/sqlalchemy/plugins/sqlalchemy_plugin.rst。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考