Litestar + SQLAlchemy Repository 实战：用 Controller 组装带分页的完整 CRUD 服务-北京尧图网络

Litestar SQLAlchemy Repository 实战用 Controller 组装带分页的完整 CRUD 服务【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本教程是 Litestar 仓库官方 SQLAlchemy Repository 系列教程docs/tutorials/repository-tutorial/的第三篇承接前两篇的数据库建模01-modelling-and-features.rst与仓库基础交互02-repository-introduction.rst重点讲解如何把仓库对象通过依赖注入接入 Litestar 的Controller并在此基础上实现一个具备列表分页、详情查询、新增、更新、删除五种能力的完整 CRUD 接口。读完本文你将掌握仓库实例的注入写法、selectinload关联加载优化、LimitOffset分页过滤器的组合方式以及异步与同步两套仓库实现的等价用法。从仓库到路由用依赖注入把 Repository 送进 Controller在前一篇教程中我们已经学会了创建AuthorRepository继承自advanced_alchemy.repository.SQLAlchemyAsyncRepository[AuthorModel]并在脚本中直接操作数据。现在要把它接入 HTTP 层最优雅的方式就是通过 Litestar 的依赖注入DI机制。首先定义一个工厂函数它返回一个AuthorRepository实例供 Controller 中的各路由注入使用。注意在这个最简示例里工厂函数只接收数据库会话db_session这一个参数没有其他配置async def provide_authors_repo(db_session: AsyncSession) - AuthorRepository: This provides the default Authors repository. return AuthorRepository(sessiondb_session)为什么函数签名里只写db_session就能拿到会话因为应用注册了SQLAlchemyInitPlugin插件见 docs/examples/sqla/sqlalchemy_async_repository.py 中SQLAlchemyInitPlugin(configsqlalchemy_config)的配置该插件会自动把数据库会话注册为名为db_session的依赖。也就是说只要使用 Litestar 的 SQLAlchemy 插件会话本身就是现成的依赖你无需手动创建或管理会话生命周期这正是文档中the session is automatically configured as a dependency的含义。按需覆盖查询语句用selectinload优化关联加载仓库默认不会给基础查询语句附加任何额外的查询选项query options但它提供了充分的灵活性可以在构造仓库时传入自定义的statement参数来覆盖默认查询。# we can optionally override the default select used for the repository to pass in # specific SQL options such as join details async def provide_author_details_repo(db_session: AsyncSession) - AuthorRepository: This provides a simple example demonstrating how to override the join options for the repository. return AuthorRepository( statementselect(AuthorModel).options(selectinload(AuthorModel.books)), sessiondb_session, )这里的关键点是selectinload(AuthorModel.books)selectinload是 SQLAlchemy 的加载策略它把关联对象通过SELECT ... IN ...模式批量加载先查询作者主记录再用主键集合执行第二条WHERE id IN (...)查询取回书籍列表最终在内存中完成组装。相比逐行触发懒加载N1 查询selectinload把查询次数压缩为两次在一对多关系一个作者对应多本书下性能优势明显。相比joinedloadJOIN方式selectinload不会因一对多笛卡尔积而膨胀主查询的行数更适合大数据量的列表场景。这个provide_author_details_repo与默认的provide_authors_repo本质上是同一个仓库类只是构造参数不同。后面我们会看到同一个依赖名authors_repo可以通过不同层级Controller 级 vs 路由级的dependencies覆盖这正是本教程展示的核心技巧之一。AuthorController五个路由的完整 CRUD 设计Controller 是 Litestar 中把一组相关路由组织在一起的载体基类定义见 litestar/controller.py。AuthorController在类上声明了dependencies并在个别路由上用dependencies参数做局部覆盖共暴露 5 个端点方法路径用途使用的仓库注入GET/authors分页列出作者provide_authors_repoController 级POST/authors新建作者provide_authors_repoController 级GET/authors/{author_id:uuid}查询单个作者含书籍provide_author_details_repo路由级PATCH/authors/{author_id:uuid}部分更新作者含书籍provide_author_details_repo路由级DELETE/authors/{author_id:uuid}删除作者provide_authors_repoController 级Controller 结构如下完整代码见 docs/examples/sqla/sqlalchemy_async_repository.pyclass AuthorController(Controller): Author CRUD dependencies {authors_repo: Provide(provide_authors_repo)} get(path/authors) async def list_authors( self, authors_repo: AuthorRepository, limit_offset: filters.LimitOffset, ) - OffsetPagination[Author]: List authors. results, total await authors_repo.list_and_count(limit_offset) type_adapter TypeAdapter(list[Author]) return OffsetPaginationAuthor, totaltotal, limitlimit_offset.limit, offsetlimit_offset.offset, ) post(path/authors) async def create_author( self, authors_repo: AuthorRepository, data: AuthorCreate, ) - Author: Create a new author. obj await authors_repo.add( AuthorModel(**data.model_dump(exclude_unsetTrue, exclude_noneTrue)), ) await authors_repo.session.commit() return Author.model_validate(obj) # we override the authors_repo to use the version that joins the Books in get(path/authors/{author_id:uuid}, dependencies{authors_repo: Provide(provide_author_details_repo)}) async def get_author( self, authors_repo: AuthorRepository, author_id: Annotated[ UUID, PathParameter( titleAuthor ID, descriptionThe author to retrieve., ), ], ) - Author: Get an existing author. obj await authors_repo.get(author_id) return Author.model_validate(obj) patch( path/authors/{author_id:uuid}, dependencies{authors_repo: Provide(provide_author_details_repo)}, ) async def update_author( self, authors_repo: AuthorRepository, data: AuthorUpdate, author_id: Annotated[ UUID, PathParameter( titleAuthor ID, descriptionThe author to update., ), ], ) - Author: Update an author. raw_obj data.model_dump(exclude_unsetTrue, exclude_noneTrue) raw_obj.update({id: author_id}) obj await authors_repo.update(AuthorModel(**raw_obj)) await authors_repo.session.commit() return Author.from_orm(obj) delete(path/authors/{author_id:uuid}) async def delete_author( self, authors_repo: AuthorRepository, author_id: Annotated[ UUID, PathParameter( titleAuthor ID, descriptionThe author to delete., ), ], ) - None: Delete a author from the system. _ await authors_repo.delete(author_id) await authors_repo.session.commit()下面逐个端点拆解其实现要点。列表端点list_and_countLimitOffset分页列表端点是数据量控制的关键。文档明确指出In our list detail endpoint, we use the pagination filter for limiting the amount of data returned, allowing us to retrieve large datasets in smaller, more manageable chunks.实现分页依赖两个配合的部件分页依赖provide_limit_offset_pagination它读取查询参数currentPage与pageSize均有ge1约束、默认值分别为 1 和 10换算成filters.LimitOffset对象并注册为limit_offset依赖def provide_limit_offset_pagination( current_page: Annotated[int, QueryParameter(namecurrentPage, ge1, requiredFalse)] 1, page_size: Annotated[int, QueryParameter(namepageSize, ge1, requiredFalse)] 10, ) - filters.LimitOffset: Add offset/limit pagination. Return type consumed by Repository.apply_limit_offset_pagination(). Parameters ---------- current_page : int LIMIT to apply to select. page_size : int OFFSET to apply to select. return filters.LimitOffset(page_size, page_size * (current_page - 1))计算逻辑很直观limit就是page_sizeoffset是page_size * (current_page - 1)第 1 页 offset 为 0。该依赖在应用级别注册Litestar(..., dependencies{limit_offset: Provide(provide_limit_offset_pagination)})。list_and_count方法它返回一个二元组(results, total)——results是当前页的模型列表total是忽略分页后的总记录数。之后用TypeAdapter(list[Author]).validate_python(results)把 ORM 对象批量转换为 Pydantic 响应模型再包装进OffsetPagination[Author]容器。OffsetPagination数据容器的定义在 litestar/pagination.py当advanced_alchemy版本大于 0.9.0 时优先使用advanced_alchemy.service.OffsetPagination包含四个字段items本页数据列表、limit每页最大条数、offset相对查询起点的偏移等价于索引、total总条数。从litestar.repository.abc._async.py的抽象定义看list_and_count的语义是返回应用过滤后的实例列表以及忽略分页的查询记录总数这正是列表接口通常需要的数据总数响应形态。创建端点add 显式commitcreate_author接收AuthorCreate仅含name、dob作为请求体通过data.model_dump(exclude_unsetTrue, exclude_noneTrue)只提取客户端确实传入的非空字段构造AuthorModel调用authors_repo.add()后手动session.commit()提交事务最后用Author.model_validate(obj)把 ORM 实例序列化返回。提示示例中仓库操作后的提交是显式的每个写操作后调用session.commit()。在真实项目中也可以使用 Litestar 的before_send处理器参考 docs/examples/sqla/plugins/sqlalchemy_async_before_send_handler.py在响应发送前统一提交从而把commit从业务代码中剥离。详情端点路由级依赖覆盖get_author的关键在于用路由级dependencies覆盖 Controller 级的同名依赖get(path/authors/{author_id:uuid}, dependencies{authors_repo: Provide(provide_author_details_repo)})也就是说同一个authors_repo参数名在GET /authors中注入的是无附加查询选项的默认仓库而在GET /authors/{author_id:uuid}中注入的是携带selectinload(AuthorModel.books)查询语句的仓库。Litestar 的依赖解析遵循越内层优先级越高的覆盖规则这种设计让你能够按端点粒度精确控制查询的加载行为而无需复制整套 CRUD 代码。路径参数author_id使用了Annotated[UUID, PathParameter(...)]注解{author_id:uuid}声明了路径中的 UUID 类型约束PathParameter还附加了 OpenAPI 文档所需的标题与描述信息。仓库的get()方法按主键查找单条记录找不到时抛出NotFoundError该行为在 litestar/repository/abc/_async.py 的check_not_found辅助方法中实现当结果为None时抛出NotFoundError(No item found when one was expected)。更新端点PATCH 语义的部分更新update_author使用patch装饰器实现部分更新语义AuthorUpdate中name与dob都是可选字段model_dump(exclude_unsetTrue, exclude_noneTrue)只保留客户端实际提交且非空的字段随后把路径参数author_id合并进更新字典raw_obj.update({id: author_id})构造出带主键的模型对象交给authors_repo.update()提交后返回更新结果。这种构造完整 ORM 对象 update()落库的写法与仓库抽象中update(data)以data上存在的属性值更新实例的语义完全一致。删除端点按主键删除delete_author直接调用authors_repo.delete(author_id)并按惯例显式commit。仓库的delete()同样会在目标不存在时抛出NotFoundError最终由 Litestar 的异常处理机制转换为对应的 HTTP 404 响应。同步仓库同一套 Controller 写法的同步版本上述示例全部基于异步实现SQLAlchemyAsyncRepositoryAsyncSession。但 Litestar 同样支持同步数据库驱动且实现完全一致——文档明确说明Litestar also supports synchronous database drivers with an identical implementation.同步版本完整代码见 docs/examples/sqla/sqlalchemy_sync_repository.py的差异点很小仓库基类换成repository.SQLAlchemySyncRepository[AuthorModel]会话类型换成sqlalchemy.orm.Session配置换成SQLAlchemySyncConfig(connection_stringsqlite:///test.sqlite)同步驱动无需aiosqlite路由处理函数不再需要async仓库方法调用去掉await例如results, total authors_repo.list_and_count(limit_offset)依赖注入处增加sync_to_threadFalseProvide(provide_authors_repo, sync_to_threadFalse)表示该依赖在事件循环线程内直接同步执行默认情况下同步依赖会被调度到线程池运行sync_to_threadFalse适合 SQLite 这类本身很快、无需跨线程的轻量驱动启动时建表由async with ... begin()改为普通with ... begin()create_all同步执行。Controller 的路由结构、分页逻辑、请求/响应模型完全复用这意味着你可以根据所连数据库驱动的性质如 SQLite 同步驱动 vs 需要异步连接的 PostgreSQL 异步驱动自由选择同步或异步仓库业务层代码几乎零成本切换。应用组装与数据库初始化无论是异步还是同步版本应用的组装方式一致把AuthorController注册进Litestar(route_handlers[...])挂载SQLAlchemyInitPlugin插件并在on_startup回调里执行metadata.create_all建表app Litestar( route_handlers[AuthorController], on_startup[on_startup], plugins[SQLAlchemyInitPlugin(configsqlalchemy_config)], dependencies{limit_offset: Provide(provide_limit_offset_pagination)}, )异步版本使用sqliteaiosqlite:///test.sqlite连接串与AsyncSessionConfig(expire_on_commitFalse)提交后不强制过期对象属性便于返回响应时继续读取 ORM 属性启动时通过sqlalchemy_config.get_engine().begin()开启事务执行base.UUIDBase.metadata.create_all。至此一个功能完整的、自带分页的 CRUD 服务就搭建完成了GET /authors分页列表、POST /authors创建、GET /authors/{author_id:uuid}详情含关联书籍加载、PATCH /authors/{author_id:uuid}部分更新、DELETE /authors/{author_id:uuid}删除。下一节教程04-repository-other.rst将在此基础上继续探讨如何扩展内置仓库以添加更多自定义能力。参考实现速查异步仓库 Controller 完整示例docs/examples/sqla/sqlalchemy_async_repository.py同步仓库 Controller 完整示例docs/examples/sqla/sqlalchemy_sync_repository.py数据库建模UUIDBase/UUIDAuditBasedocs/tutorials/repository-tutorial/01-modelling-and-features.rst仓库 CRUD 方法总览与批量操作docs/tutorials/repository-tutorial/02-repository-introduction.rst仓库抽象接口异步/同步litestar/repository/abc/_async.py过滤器类型LimitOffset等litestar/repository/filters.py分页响应容器OffsetPagination/ClassicPagination/CursorPaginationlitestar/pagination.py【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Litestar + SQLAlchemy Repository 实战：用 Controller 组装带分页的完整 CRUD 服务

相关推荐

Win11Debloat 使用指南：免费三步卸载预装应用、关闭遥测，给 Windows 11 瘦身

DenseUnet腹部多脏器分割实战：3D密集连接与解剖先验融合

DPO原理与实战：从RLHF到直接偏好优化的完整指南

更多资讯

从零构建桌面协同CRM：客户管理、工单系统与消息中心的技术实践

技能进化趋势与实战方法论：从AI协作到能力跃迁

基于Matlab的2DPSK调制解调系统仿真与误码率分析

InternVL Flash Attention配置详解：多模态训练的性能加速器

基于YOLOv5的实时人脸识别与异常行为检测系统解析

es-toolkit 的 methodOf：固定对象、后置路径的方法调用工厂函数实战指南

今日更新

基于MATLAB的CRI显色指数计算：从SPD光谱到Ra的完整流程

JSP+Servlet+MySQL博客系统源码部署与优化全攻略

蜂窝调度算法 MATLAB 实现：比例公平、轮询与最大 C/I 对比

本周精选

AI驱动的自动化专利分析平台技术解析

LeetCode-Go 题解：1252. Cells with Odd Values in a Matrix（奇数值单元格计数）双解法剖析

在线答疑系统Java毕设实战：Spring Boot前后端分离与状态流转

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？