Litestar SQLAlchemy 插件最终整合使用 SQLAlchemyPlugin 精简配置并全面回顾【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇教程是 Litestar 官方 SQLAlchemy TODO 应用系列教程的收尾章节核心解决如何用最少的样板代码把 SQLAlchemy 集成进 Litestar 应用这一实战问题。上一阶段我们同时注册了SQLAlchemyInitPlugin与SQLAlchemySerializationPlugin两个插件本阶段将用二合一的SQLAlchemyPlugin取而代之得到最终的精简版本同时完整回顾整个教程中 TODO 应用的演进路径——从手写 engine/session 生命周期管理、到依赖注入、再到序列化插件与初始化插件最终收敛为一个不足 90 行的完整异步应用。读完本文你将掌握 Advanced Alchemy 插件体系的取舍逻辑、SQLAlchemyAsyncConfig的关键配置项以及插件化集成背后的生命周期与依赖注入原理。从两个插件到一个插件配置的最终简化在整个教程中我们逐步累积了两类插件能力SQLAlchemySerializationPlugin让 Litestar 可以直接对 SQLAlchemy 模型进行请求体反序列化与响应序列化从而在处理器中直接收发模型实例见 2-serialization-plugin.rstSQLAlchemyInitPlugin在应用 lifespan 范围内自动创建并管理数据库 engine在请求范围内自动创建并管理数据库 session并通过db_session依赖注入提供给处理器见 3-init-plugin.rst。上一版代码需要同时导入并注册两个插件from advanced_alchemy.extensions.litestar import ( SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin, SQLAlchemySerializationPlugin, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[ SQLAlchemySerializationPlugin(), SQLAlchemyInitPlugin(db_config), ], )完整版本见 full_app_with_init_plugin.py。而 Advanced Alchemy 提供了SQLAlchemyPlugin作为上述两者的组合捷径。它内部同时承担初始化与序列化两类职责因此我们只需要注册一次配置即告完成。这是本阶段唯一也是最终的一次改动属于典型的收尾打磨final touches功能不变配置面显著收窄心智负担更低。最终版完整应用以下是整个教程的最终应用来自 full_app_with_plugin.py第 10、82 行是本阶段改动的高亮位置from collections.abc import AsyncGenerator from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin 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.status_codes import HTTP_409_CONFLICT class Base(DeclarativeBase): ... class TodoItem(Base): __tablename__ todo_items title: Mapped[str] mapped_column(primary_keyTrue) 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_codeHTTP_409_CONFLICT, detailstr(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(detailfTODO {todo_name!r} not found) from e async def get_todo_list(done: 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: 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 db_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///todo.sqlite, metadataBase.metadata, create_allTrue, before_send_handlerautocommit, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[SQLAlchemyPlugin(db_config)], )与上一版相比改动仅有两处导入从两个插件合并为SQLAlchemyPluginplugins列表从两项缩为一项同时新增before_send_handlerautocommit配置。其余业务逻辑保持不变。逐块解读最终应用数据模型TodoItemTodoItem表示一条 TODO 记录继承自 SQLAlchemy ORM 提供的DeclarativeBase基类此处我们自定义了Base作为统一基类。它包含两个字段以title作为主键的字符串列以及表示完成状态的布尔列doneclass Base(DeclarativeBase): ... class TodoItem(Base): __tablename__ todo_items title: Mapped[str] mapped_column(primary_keyTrue) done: Mapped[bool]得益于序列化插件处理器可以直接把TodoItem作为请求体类型和响应类型使用无需像教程第一阶段那样手工维护TodoType、TodoCollectionType类型别名和serialize_todo()转换函数对比版本见 full_app_no_plugins.py。事务依赖provide_transaction该依赖集中管理数据库事务与错误处理是教程第二阶段引入依赖注入后的核心成果。它依赖于db_session——这是由初始化插件在请求范围内自动创建并注入的 session 对象——随后通过transaction参数注入到各处理器中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_codeHTTP_409_CONFLICT, detailstr(exc), ) from exc值得注意的两点设计它把事务边界与HTTP 请求边界绑定async with db_session.begin()开启事务请求处理结束、依赖被清理时事务自动提交它把唯一性冲突统一映射为 HTTP 409 Conflict 响应任何在事务中抛出IntegrityError的操作不仅仅是新增条目都会被统一转换为ClientException从而将错误处理从各处理器中剥离覆盖范围更广。这与教程第一阶段把异常处理散落在add_item()内部的写法形成鲜明对比。查询工具函数两个异步辅助函数封装了对数据库的读取逻辑供处理器复用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(detailfTODO {todo_name!r} not found) from e async def get_todo_list(done: 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_todo_by_title()按标题精确查询单条记录使用scalar_one()并要求恰好命中一条若未找到则把 SQLAlchemy 的NoResultFound转换为 Litestar 的NotFoundException404get_todo_list()查询全部 TODO支持通过可选参数done按完成状态过滤返回模型实例列表。路由处理器TODO 的增、查、改三个处理器构成 TODO API 的完整接口第 51–69 行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要点三个处理器都通过transaction: AsyncSession参数接收注入的数据库会话——这正是dependencies{transaction: provide_transaction}在应用级注册的效果处理器按参数名自动匹配依赖add_item()直接transaction.add(data)后返回模型实例序列化插件负责把实例转换为 JSON 响应写入会在请求结束时随事务一起提交update_item()先按路径参数item_title查得既有记录再用请求体中的数据覆盖title与done更新后的对象直接作为响应返回返回值语义比教程早期版本更符合常规 API 预期add与update不再返回整个集合而是只返回被新增/更新的那一条。应用装配与插件配置最终的应用定义只有 5 行核心逻辑第 78–83 行db_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///todo.sqlite, metadataBase.metadata, create_allTrue, before_send_handlerautocommit, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[SQLAlchemyPlugin(db_config)], )SQLAlchemyAsyncConfig是本阶段值得展开的关键配置对象各参数作用如下配置项取值示例作用connection_stringsqliteaiosqlite:///todo.sqlite指定异步数据库连接串这里使用 aiosqlite 驱动连接本地 SQLite 文件生产环境可替换为 PostgreSQL/MySQL 等异步驱动连接串metadataBase.metadata提供 ORM 元数据供插件在create_allTrue时按模型建表create_allTrue应用启动时自动调用Base.metadata.create_all创建缺失的表表已存在则跳过before_send_handlerautocommit响应发送前自动提交事务的内置处理策略避免在处理器中手工commit()插件侧SQLAlchemyPlugin(db_config)是本次整合的落点它等价于同时注册SQLAlchemySerializationPlugin()与SQLAlchemyInitPlugin(db_config)一次性获得模型序列化能力与 engine/session 生命周期管理能力。此后engine 生命周期由插件在应用 lifespan 内接管对比第一阶段需手写db_connection()生命周期上下文管理器并在app.state.engine中存取 engine参见 0-introduction.rstsession 生命周期由插件在请求范围内接管并通过db_session依赖注入暴露给用户依赖表结构初始化由create_allTrue自动完成无需在 lifespan 中调用conn.run_sync(Base.metadata.create_all)。从源码结构可以推断SQLAlchemyPlugin承担的是组合根角色它在应用初始化时同时装配初始化插件与序列化插件的钩子对外呈现为单一入口这正是本教程把两类能力二合一、收敛配置面的实现依据。教程全程回顾TODO 应用的五次演进整个 SQLAlchemy 教程目录见 index.rst以 TODO 应用为载体展示了 Litestar 与 Advanced Alchemy 集成的渐进式优化路径最终版是前四步成果的合流阶段文档章节核心改进对应示例文件1. 基线0-introduction.rst按 SQLAlchemy 官方文档风格手写 engine/session 管理lifespan 上下文管理器 应用状态 手工序列化full_app_no_plugins.py2. 依赖注入1-provide-session-with-di.rst用provide_transaction()依赖集中创建 session、开启事务并统一处理IntegrityErrorfull_app_with_session_di.py3. 序列化插件2-serialization-plugin.rst引入SQLAlchemySerializationPlugin处理器直接收发模型实例删除类型别名与serialize_todo()full_app_with_serialization_plugin.py4. 初始化插件3-init-plugin.rst引入SQLAlchemyInitPlugin删除手写 lifespanengine 与 session 交给插件管理新增db_session依赖full_app_with_init_plugin.py5. 二合一收尾本文用SQLAlchemyPlugin合并两个插件新增before_send_handlerautocommit配置面收敛到最小full_app_with_plugin.py回顾整个系列可以提炼出四条贯穿始终的设计主线资源管理上移从处理器内手写 session → 依赖注入统一提供 → 插件在 lifespan/请求作用域自动托管资源获取与释放的样板代码逐步消失序列化能力内建从手工维护 DTO 别名与转换函数 → 序列化插件直接理解 ORM 模型处理器签名更接近业务本身错误处理收敛从各处理器分别捕获IntegrityError→ 集中在事务依赖中统一映射为 409覆盖面更广、代码更 DRY配置单一入口从两个插件并列 →SQLAlchemyPlugin组合业务代码只需关心SQLAlchemyAsyncConfig一处配置。快速上手与运行前提要复现本教程应用需要安装 Advanced Alchemy 及异步 SQLite 驱动官方推荐两种方式# 方式一直接安装 Advanced Alchemy含 aiosqlite 依赖组 pip install advanced-alchemy[aiosqlite] # 方式二通过 Litestar 的 sqlalchemy 扩展安装 pip install litestar[standard,sqlalchemy] aiosqlite安装后直接运行 full_app_with_plugin.py应用会基于sqliteaiosqlite:///todo.sqlite自动创建todo.sqlite与todo_items表然后即可通过GET /、POST /、PUT /{item_title:str}三个接口完成 TODO 的查询、新增与更新。需要特别说明的兼容性前提SQLAlchemy 支持在 Litestar 中由 Advanced Alchemy 这一第一方库提供所有导入应使用advanced_alchemy.extensions.litestar命名空间而不是已废弃的litestar.contrib.sqlalchemy或litestar.plugins.sqlalchemy模块见 index.rst。本文所有示例均为异步风格AsyncSession aiosqlite仓库中同时提供同步 SQLAlchemy 版本sqlalchemy_sync.py与异步版本sqlalchemy_async.py可供对照若需要更多配置项细节可进一步查阅 Advanced Alchemy 官方文档。小结最终版的 TODO 应用证明了插件化集成的收益业务代码只保留模型、依赖、查询工具与路由四个层次数据库的 engine 生命周期、session 生命周期、建表与序列化全部由SQLAlchemyPlugin背后的 Advanced Alchemy 接管。从手工样板到插件组合代码量减少的同时资源管理、错误处理与类型边界反而更加清晰——这正是 Litestar 生态约定优于配置风格的典型体现。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Litestar SQLAlchemy 插件最终整合使用 SQLAlchemyPlugin 精简配置并全面回顾【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本篇教程是 Litestar 官方 SQLAlchemy TODO 应用系列教程的收尾章节核心解决如何用最少的样板代码把 SQLAlchemy 集成进 Litestar 应用这一实战问题。上一阶段我们同时注册了SQLAlchemyInitPlugin与SQLAlchemySerializationPlugin两个插件本阶段将用二合一的SQLAlchemyPlugin取而代之得到最终的精简版本同时完整回顾整个教程中 TODO 应用的演进路径——从手写 engine/session 生命周期管理、到依赖注入、再到序列化插件与初始化插件最终收敛为一个不足 90 行的完整异步应用。读完本文你将掌握 Advanced Alchemy 插件体系的取舍逻辑、SQLAlchemyAsyncConfig的关键配置项以及插件化集成背后的生命周期与依赖注入原理。从两个插件到一个插件配置的最终简化在整个教程中我们逐步累积了两类插件能力SQLAlchemySerializationPlugin让 Litestar 可以直接对 SQLAlchemy 模型进行请求体反序列化与响应序列化从而在处理器中直接收发模型实例见 2-serialization-plugin.rstSQLAlchemyInitPlugin在应用 lifespan 范围内自动创建并管理数据库 engine在请求范围内自动创建并管理数据库 session并通过db_session依赖注入提供给处理器见 3-init-plugin.rst。上一版代码需要同时导入并注册两个插件from advanced_alchemy.extensions.litestar import ( SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin, SQLAlchemySerializationPlugin, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[ SQLAlchemySerializationPlugin(), SQLAlchemyInitPlugin(db_config), ], )完整版本见 full_app_with_init_plugin.py。而 Advanced Alchemy 提供了SQLAlchemyPlugin作为上述两者的组合捷径。它内部同时承担初始化与序列化两类职责因此我们只需要注册一次配置即告完成。这是本阶段唯一也是最终的一次改动属于典型的收尾打磨final touches功能不变配置面显著收窄心智负担更低。最终版完整应用以下是整个教程的最终应用来自 full_app_with_plugin.py第 10、82 行是本阶段改动的高亮位置from collections.abc import AsyncGenerator from advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin 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.status_codes import HTTP_409_CONFLICT class Base(DeclarativeBase): ... class TodoItem(Base): __tablename__ todo_items title: Mapped[str] mapped_column(primary_keyTrue) 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_codeHTTP_409_CONFLICT, detailstr(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(detailfTODO {todo_name!r} not found) from e async def get_todo_list(done: 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: 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 db_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///todo.sqlite, metadataBase.metadata, create_allTrue, before_send_handlerautocommit, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[SQLAlchemyPlugin(db_config)], )与上一版相比改动仅有两处导入从两个插件合并为SQLAlchemyPluginplugins列表从两项缩为一项同时新增before_send_handlerautocommit配置。其余业务逻辑保持不变。逐块解读最终应用数据模型TodoItemTodoItem表示一条 TODO 记录继承自 SQLAlchemy ORM 提供的DeclarativeBase基类此处我们自定义了Base作为统一基类。它包含两个字段以title作为主键的字符串列以及表示完成状态的布尔列doneclass Base(DeclarativeBase): ... class TodoItem(Base): __tablename__ todo_items title: Mapped[str] mapped_column(primary_keyTrue) done: Mapped[bool]得益于序列化插件处理器可以直接把TodoItem作为请求体类型和响应类型使用无需像教程第一阶段那样手工维护TodoType、TodoCollectionType类型别名和serialize_todo()转换函数对比版本见 full_app_no_plugins.py。事务依赖provide_transaction该依赖集中管理数据库事务与错误处理是教程第二阶段引入依赖注入后的核心成果。它依赖于db_session——这是由初始化插件在请求范围内自动创建并注入的 session 对象——随后通过transaction参数注入到各处理器中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_codeHTTP_409_CONFLICT, detailstr(exc), ) from exc值得注意的两点设计它把事务边界与HTTP 请求边界绑定async with db_session.begin()开启事务请求处理结束、依赖被清理时事务自动提交它把唯一性冲突统一映射为 HTTP 409 Conflict 响应任何在事务中抛出IntegrityError的操作不仅仅是新增条目都会被统一转换为ClientException从而将错误处理从各处理器中剥离覆盖范围更广。这与教程第一阶段把异常处理散落在add_item()内部的写法形成鲜明对比。查询工具函数两个异步辅助函数封装了对数据库的读取逻辑供处理器复用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(detailfTODO {todo_name!r} not found) from e async def get_todo_list(done: 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_todo_by_title()按标题精确查询单条记录使用scalar_one()并要求恰好命中一条若未找到则把 SQLAlchemy 的NoResultFound转换为 Litestar 的NotFoundException404get_todo_list()查询全部 TODO支持通过可选参数done按完成状态过滤返回模型实例列表。路由处理器TODO 的增、查、改三个处理器构成 TODO API 的完整接口第 51–69 行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要点三个处理器都通过transaction: AsyncSession参数接收注入的数据库会话——这正是dependencies{transaction: provide_transaction}在应用级注册的效果处理器按参数名自动匹配依赖add_item()直接transaction.add(data)后返回模型实例序列化插件负责把实例转换为 JSON 响应写入会在请求结束时随事务一起提交update_item()先按路径参数item_title查得既有记录再用请求体中的数据覆盖title与done更新后的对象直接作为响应返回返回值语义比教程早期版本更符合常规 API 预期add与update不再返回整个集合而是只返回被新增/更新的那一条。应用装配与插件配置最终的应用定义只有 5 行核心逻辑第 78–83 行db_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///todo.sqlite, metadataBase.metadata, create_allTrue, before_send_handlerautocommit, ) app Litestar( [get_list, add_item, update_item], dependencies{transaction: provide_transaction}, plugins[SQLAlchemyPlugin(db_config)], )SQLAlchemyAsyncConfig是本阶段值得展开的关键配置对象各参数作用如下配置项取值示例作用connection_stringsqliteaiosqlite:///todo.sqlite指定异步数据库连接串这里使用 aiosqlite 驱动连接本地 SQLite 文件生产环境可替换为 PostgreSQL/MySQL 等异步驱动连接串metadataBase.metadata提供 ORM 元数据供插件在create_allTrue时按模型建表create_allTrue应用启动时自动调用Base.metadata.create_all创建缺失的表表已存在则跳过before_send_handlerautocommit响应发送前自动提交事务的内置处理策略避免在处理器中手工commit()插件侧SQLAlchemyPlugin(db_config)是本次整合的落点它等价于同时注册SQLAlchemySerializationPlugin()与SQLAlchemyInitPlugin(db_config)一次性获得模型序列化能力与 engine/session 生命周期管理能力。此后engine 生命周期由插件在应用 lifespan 内接管对比第一阶段需手写db_connection()生命周期上下文管理器并在app.state.engine中存取 engine参见 0-introduction.rstsession 生命周期由插件在请求范围内接管并通过db_session依赖注入暴露给用户依赖表结构初始化由create_allTrue自动完成无需在 lifespan 中调用conn.run_sync(Base.metadata.create_all)。从源码结构可以推断SQLAlchemyPlugin承担的是组合根角色它在应用初始化时同时装配初始化插件与序列化插件的钩子对外呈现为单一入口这正是本教程把两类能力二合一、收敛配置面的实现依据。教程全程回顾TODO 应用的五次演进整个 SQLAlchemy 教程目录见 index.rst以 TODO 应用为载体展示了 Litestar 与 Advanced Alchemy 集成的渐进式优化路径最终版是前四步成果的合流阶段文档章节核心改进对应示例文件1. 基线0-introduction.rst按 SQLAlchemy 官方文档风格手写 engine/session 管理lifespan 上下文管理器 应用状态 手工序列化full_app_no_plugins.py2. 依赖注入1-provide-session-with-di.rst用provide_transaction()依赖集中创建 session、开启事务并统一处理IntegrityErrorfull_app_with_session_di.py3. 序列化插件2-serialization-plugin.rst引入SQLAlchemySerializationPlugin处理器直接收发模型实例删除类型别名与serialize_todo()full_app_with_serialization_plugin.py4. 初始化插件3-init-plugin.rst引入SQLAlchemyInitPlugin删除手写 lifespanengine 与 session 交给插件管理新增db_session依赖full_app_with_init_plugin.py5. 二合一收尾本文用SQLAlchemyPlugin合并两个插件新增before_send_handlerautocommit配置面收敛到最小full_app_with_plugin.py回顾整个系列可以提炼出四条贯穿始终的设计主线资源管理上移从处理器内手写 session → 依赖注入统一提供 → 插件在 lifespan/请求作用域自动托管资源获取与释放的样板代码逐步消失序列化能力内建从手工维护 DTO 别名与转换函数 → 序列化插件直接理解 ORM 模型处理器签名更接近业务本身错误处理收敛从各处理器分别捕获IntegrityError→ 集中在事务依赖中统一映射为 409覆盖面更广、代码更 DRY配置单一入口从两个插件并列 →SQLAlchemyPlugin组合业务代码只需关心SQLAlchemyAsyncConfig一处配置。快速上手与运行前提要复现本教程应用需要安装 Advanced Alchemy 及异步 SQLite 驱动官方推荐两种方式# 方式一直接安装 Advanced Alchemy含 aiosqlite 依赖组 pip install advanced-alchemy[aiosqlite] # 方式二通过 Litestar 的 sqlalchemy 扩展安装 pip install litestar[standard,sqlalchemy] aiosqlite安装后直接运行 full_app_with_plugin.py应用会基于sqliteaiosqlite:///todo.sqlite自动创建todo.sqlite与todo_items表然后即可通过GET /、POST /、PUT /{item_title:str}三个接口完成 TODO 的查询、新增与更新。需要特别说明的兼容性前提SQLAlchemy 支持在 Litestar 中由 Advanced Alchemy 这一第一方库提供所有导入应使用advanced_alchemy.extensions.litestar命名空间而不是已废弃的litestar.contrib.sqlalchemy或litestar.plugins.sqlalchemy模块见 index.rst。本文所有示例均为异步风格AsyncSession aiosqlite仓库中同时提供同步 SQLAlchemy 版本sqlalchemy_sync.py与异步版本sqlalchemy_async.py可供对照若需要更多配置项细节可进一步查阅 Advanced Alchemy 官方文档。小结最终版的 TODO 应用证明了插件化集成的收益业务代码只保留模型、依赖、查询工具与路由四个层次数据库的 engine 生命周期、session 生命周期、建表与序列化全部由SQLAlchemyPlugin背后的 Advanced Alchemy 接管。从手工样板到插件组合代码量减少的同时资源管理、错误处理与类型边界反而更加清晰——这正是 Litestar 生态约定优于配置风格的典型体现。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
