Litestar 集成 SQLAlchemy 完整实战模型基类、Repository 与三大 Advanced-Alchemy 插件【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar本文基于 Litestar 官方文档 SQLAlchemy 章节 及其配套示例系统讲解如何在 Litestar 应用中使用 SQLAlchemy 2.0从预置DeclarativeBase模型基类、通用 CRUD Repository到SQLAlchemyPlugin、SQLAlchemyInitPlugin、SQLAlchemySerializationPlugin三个插件的职责与配置参数。读完本文你可以独立完成“模型定义 → Repository 接入 Controller → 插件装配 → 序列化托管”的完整数据库层搭建。一、安装与适用前提SQLAlchemy 支持通过安装附加包直接启用pip install litestar[sqlalchemy]该集成基于Advanced-Alchemy库仅兼容 SQLAlchemy 2.0源码中使用Mapped/mapped_column等 2.0 风格 API 也印证了这一点。文档明确列出的经过测试支持的数据库后端包括SQLiteaiosqlite或标准库sqlite3Postgresasyncpg或psycopg3异步/同步均可MySQLasyncmyOracleoracledbGoogle Spannerspanner-sqlalchemy、DuckDBduckdb_engine、Microsoft SQL Serverpyodbc二、定义模型预置 DeclarativeBase 与字段增强要使用内置的SQLAlchemyAsyncRepository首先需要基于 Advanced-Alchemy 提供的DeclarativeBase实现来声明模型。共有四个基类可选基类主键审计列advanced_alchemy.base.UUIDBaseUUID无advanced_alchemy.base.UUIDAuditBaseUUIDcreated_at/updated_atadvanced_alchemy.base.BigIntBaseBigInteger无advanced_alchemy.base.BigIntAuditBaseBigIntegercreated_at/updated_at这些基类在原生 SQLAlchemy 模型之上提供了若干增强UUID 的跨方言存储在 Postgres 等支持原生 UUID/GUID 类型的数据库上id字段使用原生类型在不支持 UUID 类型的引擎上则退化为 16 字节的BYTES/RAW字段存储自动表名从类名自动生成 snake-case 表名也可以显式声明__tablename__覆盖优化的 JSON 列类型模型中出现的 PydanticBaseModel或Dict类字段会被映射为跨方言的 JSON 类型——Postgres 上是JSONBOracle 上是有 JSON 校验约束的VARCHAR/BYTES其他方言上是标准JSON。完整的模型定义示例来源sqlalchemy_declarative_models.pyfrom __future__ import annotations import uuid from datetime import date from uuid import UUID from advanced_alchemy.extensions.litestar import ( AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyPlugin, base, ) from sqlalchemy import ForeignKey, func, select from sqlalchemy.orm import Mapped, mapped_column, relationship from litestar import Litestar, get from litestar.di import NamedDependency # UUIDBase 包含一个 UUID 主键id class Author(base.UUIDBase): __tablename__ author name: Mapped[str] dob: Mapped[date] books: Mapped[list[Book]] relationship(back_populatesauthor, lazyselectin) # UUIDAuditBase 在 UUID 主键之外额外包含 created_at 与 updated_at class Book(base.UUIDAuditBase): __tablename__ book title: Mapped[str] author_id: Mapped[UUID] mapped_column(ForeignKey(author.id)) author: Mapped[Author] relationship(lazyjoined, innerjoinTrue, viewonlyTrue) session_config AsyncSessionConfig(expire_on_commitFalse) sqlalchemy_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///test.sqlite, session_configsession_config, create_allTrue, # 应用启动时自动建表 ) async def on_startup(app: Litestar) - None: 若无数据则写入示例数据。 async with sqlalchemy_config.get_session() as session: count (await session.execute( select(func.count()).select_from(Author) )).scalar() if not count: author_id uuid.uuid4() session.add(Author(nameStephen King, dobdate(1954, 9, 21), idauthor_id)) session.add(Book(titleIt, author_idauthor_id)) await session.commit() get(path/authors) async def get_authors( db_session: NamedDependency, db_engine: NamedDependency ) - list[Author]: 通过注入的 session 与 engine 与数据库交互。 return list(await db_session.scalars(select(Author))) app Litestar( route_handlers[get_authors], on_startup[on_startup], debugTrue, plugins[SQLAlchemyPlugin(configsqlalchemy_config)], )注意SQLAlchemyAsyncConfig的两个关键参数connection_string使用 SQLAlchemy 的 URL 语法同步引擎如sqlite:///test.sqlite异步引擎需使用sqliteaiosqlite、postgresqlasyncpg等带异步驱动前缀的形式create_allTrue会在启动阶段自动执行建表适合演示与测试场景。三、Repository内置的通用 CRUD 层Advanced-Alchemy 提供泛型同步/异步 Repositoryadvanced_alchemy.repository.SQLAlchemyAsyncRepository/SQLAlchemySyncRepository覆盖 select、insert、update、delete 四类操作并且针对批量插入、更新、删除实现了优化方法尽可能使用 SQLAlchemy 的lambda_stmt机制生成单条高效 SQL内建 count、分页、排序以及LIKE、IN、日期 before/after 等过滤能力过滤条件定义在filters模块中如LimitOffset。3.1 异步 Repository 完整示例以下示例来自 sqlalchemy_async_repository.py展示了“模型 Pydantic Schema Repository Controller 分页依赖”的完整闭环from advanced_alchemy.extensions.litestar import ( AsyncSessionConfig, SQLAlchemyAsyncConfig, SQLAlchemyInitPlugin, base, filters, repository, ) from pydantic import BaseModel as _BaseModel from pydantic import TypeAdapter from sqlalchemy import ForeignKey, select from sqlalchemy.orm import Mapped, mapped_column, relationship, selectinload from litestar import Litestar, get from litestar.controller import Controller from litestar.di import Provide from litestar.handlers.http_handlers.decorators import delete, patch, post from litestar.pagination import OffsetPagination from litestar.params import PathParameter, QueryParameter class BaseModel(_BaseModel): 扩展 Pydantic BaseModel 以启用 ORM 模式。 model_config {from_attributes: True} class AuthorModel(base.UUIDBase): # 可选显式指定表名否则从类名自动生成 __tablename__ author name: Mapped[str] dob: Mapped[date | None] books: Mapped[list[BookModel]] relationship(back_populatesauthor, lazynoload) class BookModel(base.UUIDAuditBase): __tablename__ book title: Mapped[str] author_id: Mapped[UUID] mapped_column(ForeignKey(author.id)) author: Mapped[AuthorModel] relationship(lazyjoined, innerjoinTrue, viewonlyTrue) class Author(BaseModel): id: UUID | None name: str dob: date | None None class AuthorCreate(BaseModel): name: str dob: date | None None class AuthorUpdate(BaseModel): name: str | None None dob: date | None None class AuthorRepository(repository.SQLAlchemyAsyncRepository[AuthorModel]): Author repository。 model_type AuthorModel async def provide_authors_repo(db_session) - AuthorRepository: 提供默认的 Author repository。 return AuthorRepository(sessiondb_session) # 可通过覆写默认 select 传入 join 选项 async def provide_author_details_repo(db_session) - AuthorRepository: return AuthorRepository( statementselect(AuthorModel).options(selectinload(AuthorModel.books)), sessiondb_session, ) 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: 从查询参数构造分页条件供 repository.apply_limit_offset_pagination() 使用。 return filters.LimitOffset(page_size, page_size * (current_page - 1)) class 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_and_count 一次返回 (当前页数据, 总数)。 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: obj await authors_repo.add(AuthorModel(**data.model_dump(exclude_unsetTrue, exclude_noneTrue))) await authors_repo.session.commit() return Author.model_validate(obj) # 该路由覆写 authors_repo使用预加载 Book 的版本 get(path/authors/{author_id:uuid}, dependencies{authors_repo: Provide(provide_author_details_repo)}) async def get_author(self, authors_repo: AuthorRepository, author_id: UUID) - 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: UUID) - 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.model_validate(obj) delete(path/authors/{author_id:uuid}) async def delete_author(self, authors_repo: AuthorRepository, author_id: UUID) - None: _ await authors_repo.delete(author_id) await authors_repo.session.commit() session_config AsyncSessionConfig(expire_on_commitFalse) sqlalchemy_config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///test.sqlite, session_configsession_config ) # 创建 db_session 依赖 sqlalchemy_plugin SQLAlchemyInitPlugin(configsqlalchemy_config) async def on_startup() - None: 初始化数据库。 async with sqlalchemy_config.get_engine().begin() as conn: await conn.run_sync(base.UUIDBase.metadata.create_all) app Litestar( route_handlers[AuthorController], on_startup[on_startup], plugins[sqlalchemy_plugin], dependencies{limit_offset: Provide(provide_limit_offset_pagination)}, )几个值得注意的实现细节Repository 只需声明model_type即可获得add/get/update/delete/list_and_count等完整方法集通过构造参数覆写默认语句AuthorRepository(statementselect(...).options(selectinload(...)))可以针对单个 repository 实例注入预加载选项而 Controller 级dependencies覆写则让“详情接口”使用预加载版本——这避免了 N1 查询分页走filters.LimitOffsetlist_and_count(limit_offset)返回(items, total)直接喂给 Litestar 自带的OffsetPagination泛型模型同步场景使用SQLAlchemySyncRepository结构完全一致见 sqlalchemy_sync_repository.py。更完整的 repository 用法CRUD、批量操作、repository 扩展可参考示例 sqlalchemy_repository_crud.py、sqlalchemy_repository_bulk_operations.py、sqlalchemy_repository_extension.py以及 Repository 教程。四、SQLAlchemyPlugin一站式集成advanced_alchemy.extensions.litestar.SQLAlchemyPlugin组合了 Init Plugin 与 Serialization Plugin 的全部能力是功能最完整的入口。下面用 SQLite 演示一个最小完整应用来源sqlalchemy_async_plugin_example.pyfrom advanced_alchemy.extensions.litestar import SQLAlchemyAsyncConfig, SQLAlchemyPlugin from sqlalchemy import select from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from litestar import Litestar, post from litestar.di import NamedDependency 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, db_session: NamedDependency) - Sequence[TodoItem]: async with db_session.begin(): db_session.add(data) return (await db_session.execute(select(TodoItem))).scalars().all() config SQLAlchemyAsyncConfig( connection_stringsqliteaiosqlite:///todo_async.sqlite, create_allTrue, metadataBase.metadata, # 指定要建表的 metadata ) plugin SQLAlchemyPlugin(configconfig) app Litestar(route_handlers[add_item], plugins[plugin])若需要启动时建表/初始化数据可以像前文on_startup那样使用config.get_engine()/config.get_session()显式初始化。文档特别强调示例中“先 drop 再 create”仅为了可重复运行生产环境不应这样做——生产应使用迁移工具管理表结构。同步版本见 sqlalchemy_sync_plugin_example.py。运行方式litestar run curl -X POST -H Content-Type: application/json \ -d {title: Your Todo Title, done: false} http://localhost:8000/五、SQLAlchemyInitPlugin依赖注入与会话生命周期如果你只需要数据库基础设施而不需要自动序列化使用SQLAlchemyInitPlugin。它为应用提供四项能力通过依赖注入暴露 engine 与 session默认键名为db_engine/db_session在应用 state 中管理 engine 与 session factory配置一个before_send钩子响应发送前默认关闭 session 并将其从连接 scope 中移除向 signature namespace 注入相关名称辅助解析注解类型。最简依赖注入示例sqlalchemy_async_dependencies.pypost(/) async def handler( db_session: NamedDependency, db_engine: NamedDependency ) - tuple[int, int]: one (await db_session.scalars(select(literal(1)))).one() async with db_engine.begin() as conn: two (await conn.scalars(select(literal(2)))).one() return one, two config SQLAlchemyAsyncConfig(connection_stringsqliteaiosqlite:///async.sqlite) plugin SQLAlchemyInitPlugin(configconfig) app Litestar(route_handlers[handler], plugins[plugin])注入的 session/engine 与普通依赖一样可以被其他依赖函数再次注入。5.1 重命名依赖键通过在配置对象上设置engine_dependency_key与session_dependency_key两个属性可以改变 engine 和 session 绑定的依赖名避免与业务依赖冲突。5.2 自定义 before_send 处理器默认before_send处理器只负责关闭 session 并清理 scope。若希望“成功则 commit、失败则 rollback”可以替换before_send_handler属性——Advanced-Alchemy 提供了一个带自动提交/回滚行为的替代处理器见 sqlalchemy_async_before_send_handler.py 与同步版 sqlalchemy_sync_before_send_handler.py。5.3 engine 与 session 配置SQLAlchemyAsyncConfig/SQLAlchemySyncConfig均有engine_config属性其类型为advanced_alchemy.extensions.litestar.EngineConfig暴露 SQLAlchemy engine 的全部配置选项连接池、回滚策略、echo 等两者的session_config属性分别是AsyncSessionConfig或SyncSessionConfig实例暴露 SQLAlchemy session 的全部配置选项如前文用到的expire_on_commitFalse用于避免响应序列化时触发“detached instance”访问异常配置对象本身还暴露若干控制插件行为的选项完整参数请查阅 Advanced-Alchemy 的 reference 文档。仅用 Init Plugin 时Handler 内部需要自行在 ORM 对象与可序列化结构之间转换SQLAlchemySerializationPlugin正是为消除这类样板代码而设计的。六、SQLAlchemySerializationPlugin自动 DTO 与字段标记SQLAlchemySerializationPlugin不接收任何参数实例化后直接传给应用即可来源sqlalchemy_async_serialization_plugin.pyfrom 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()])工作原理应用注册阶段凡是 handler 的data参数或返回注解为 SQLAlchemy 模型或模型集合、且未显式指定 DTO 类的插件会为它们生成SQLAlchemyDTO。因此下面两种写法功能等价挂载SQLAlchemySerializationPlugin零声明显式定义 DTO 类见 sqlalchemy_async_serialization_dto.py。字段标记控制出入数据既然插件只是“替 handler 定义 DTO”就可以用 DTO 的字段标记机制mark控制哪些字段允许进出。示例 sqlalchemy_async_serialization_plugin_marking_fields.py 中模型新增了一个super_secret_value属性并在 handler 中赋值但该字段被标记为 private 后序列化响应中不会出现它——这是把敏感字段如内部状态、凭据挡在 API 边界之外的标准做法。七、三个插件的选型关系与延伸阅读插件职责适用场景SQLAlchemyPluginInit Serialization 全量能力最常见选择快速接入SQLAlchemyInitPlugin引擎/会话的依赖注入、state 管理、before_send、namespace手动管理序列化只需数据库基础设施SQLAlchemySerializationPlugin为 ORM 模型注解自动生成 DTO已有会话管理只想免除 DTO 样板代码从源码结构看SQLAlchemyPlugin是对后两者的组合封装三者可以按需独立挂载。仓库中还提供了一组“同一应用的四种装配方式”对照示例便于横向比较样板代码量的差异full_app_no_plugins.py、full_app_with_init_plugin.py、full_app_with_serialization_plugin.py 与 full_app_with_plugin.py。关键文档与示例路径一览章节索引docs/usage/databases/sqlalchemy/index.rst模型与 Repositorydocs/usage/databases/sqlalchemy/models_and_repository.rst插件总览docs/usage/databases/sqlalchemy/plugins/index.rst各插件文档SQLAlchemy Plugin、Init Plugin、Serialization Plugin同步版示例sqlalchemy_sync_plugin_example.py、sqlalchemy_sync_dependencies.py需要强调的是适用前提以上所有示例均以 SQLiteaiosqlite/标准库演示切换到 Postgres 等生产数据库时只需更换connection_string的驱动前缀如postgresqlasyncpg://...模型、Repository 与插件代码均可保持不变而表结构变更请交由迁移工具而非create_all处理。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
