Litestar 集成 SQLAlchemy 完整實戰(zhàn)模型基類、Repository 與三大 Advanced-Alchemy 插件【免費(fèi)下載鏈接】litestarLight, flexible and extensible ASGI framework | Built to scale項目地址: https://gitcode.com/GitHub_Trending/li/litestar本文基于 Litestar 官方文檔 SQLAlchemy 章節(jié) 及其配套示例系統(tǒng)講解如何在 Litestar 應(yīng)用中使用 SQLAlchemy 2.0從預(yù)置DeclarativeBase模型基類、通用 CRUD Repository到SQLAlchemyPlugin、SQLAlchemyInitPlugin、SQLAlchemySerializationPlugin三個插件的職責(zé)與配置參數(shù)。讀完本文你可以獨(dú)立完成“模型定義 → Repository 接入 Controller → 插件裝配 → 序列化托管”的完整數(shù)據(jù)庫層搭建。一、安裝與適用前提SQLAlchemy 支持通過安裝附加包直接啟用pip install litestar[sqlalchemy]該集成基于Advanced-Alchemy庫僅兼容 SQLAlchemy 2.0源碼中使用Mapped/mapped_column等 2.0 風(fēng)格 API 也印證了這一點(diǎn)。文檔明確列出的經(jīng)過測試支持的數(shù)據(jù)庫后端包括SQLiteaiosqlite或標(biāo)準(zhǔn)庫sqlite3Postgresasyncpg或psycopg3異步/同步均可MySQLasyncmyOracleoracledbGoogle Spannerspanner-sqlalchemy、DuckDBduckdb_engine、Microsoft SQL Serverpyodbc二、定義模型預(yù)置 DeclarativeBase 與字段增強(qiáng)要使用內(nèi)置的SQLAlchemyAsyncRepository首先需要基于 Advanced-Alchemy 提供的DeclarativeBase實現(xiàn)來聲明模型。共有四個基類可選基類主鍵審計列advanced_alchemy.base.UUIDBaseUUID無advanced_alchemy.base.UUIDAuditBaseUUIDcreated_at/updated_atadvanced_alchemy.base.BigIntBaseBigInteger無advanced_alchemy.base.BigIntAuditBaseBigIntegercreated_at/updated_at這些基類在原生 SQLAlchemy 模型之上提供了若干增強(qiáng)UUID 的跨方言存儲在 Postgres 等支持原生 UUID/GUID 類型的數(shù)據(jù)庫上id字段使用原生類型在不支持 UUID 類型的引擎上則退化為 16 字節(jié)的BYTES/RAW字段存儲自動表名從類名自動生成 snake-case 表名也可以顯式聲明__tablename__覆蓋優(yōu)化的 JSON 列類型模型中出現(xiàn)的 PydanticBaseModel或Dict類字段會被映射為跨方言的 JSON 類型——Postgres 上是JSONBOracle 上是有 JSON 校驗約束的VARCHAR/BYTES其他方言上是標(biāo)準(zhǔn)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, # 應(yīng)用啟動時自動建表 ) async def on_startup(app: Litestar) - None: 若無數(shù)據(jù)則寫入示例數(shù)據(jù)。 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 與數(shù)據(jù)庫交互。 return list(await db_session.scalars(select(Author))) app Litestar( route_handlers[get_authors], on_startup[on_startup], debugTrue, plugins[SQLAlchemyPlugin(configsqlalchemy_config)], )注意SQLAlchemyAsyncConfig的兩個關(guān)鍵參數(shù)connection_string使用 SQLAlchemy 的 URL 語法同步引擎如sqlite:///test.sqlite異步引擎需使用sqliteaiosqlite、postgresqlasyncpg等帶異步驅(qū)動前綴的形式create_allTrue會在啟動階段自動執(zhí)行建表適合演示與測試場景。三、Repository內(nèi)置的通用 CRUD 層Advanced-Alchemy 提供泛型同步/異步 Repositoryadvanced_alchemy.repository.SQLAlchemyAsyncRepository/SQLAlchemySyncRepository覆蓋 select、insert、update、delete 四類操作并且針對批量插入、更新、刪除實現(xiàn)了優(yōu)化方法盡可能使用 SQLAlchemy 的lambda_stmt機(jī)制生成單條高效 SQL內(nèi)建 count、分頁、排序以及LIKE、IN、日期 before/after 等過濾能力過濾條件定義在filters模塊中如LimitOffset。3.1 異步 Repository 完整示例以下示例來自 sqlalchemy_async_repository.py展示了“模型 Pydantic Schema Repository Controller 分頁依賴”的完整閉環(huán)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): 擴(kuò)展 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: 提供默認(rèn)的 Author repository。 return AuthorRepository(sessiondb_session) # 可通過覆寫默認(rèn) 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: 從查詢參數(shù)構(gòu)造分頁條件供 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 一次返回 (當(dāng)前頁數(shù)據(jù), 總數(shù))。 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使用預(yù)加載 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 ) # 創(chuàng)建 db_session 依賴 sqlalchemy_plugin SQLAlchemyInitPlugin(configsqlalchemy_config) async def on_startup() - None: 初始化數(shù)據(jù)庫。 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)}, )幾個值得注意的實現(xiàn)細(xì)節(jié)Repository 只需聲明model_type即可獲得add/get/update/delete/list_and_count等完整方法集通過構(gòu)造參數(shù)覆寫默認(rèn)語句AuthorRepository(statementselect(...).options(selectinload(...)))可以針對單個 repository 實例注入預(yù)加載選項而 Controller 級dependencies覆寫則讓“詳情接口”使用預(yù)加載版本——這避免了 N1 查詢分頁走filters.LimitOffsetlist_and_count(limit_offset)返回(items, total)直接喂給 Litestar 自帶的OffsetPagination泛型模型同步場景使用SQLAlchemySyncRepository結(jié)構(gòu)完全一致見 sqlalchemy_sync_repository.py。更完整的 repository 用法CRUD、批量操作、repository 擴(kuò)展可參考示例 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 演示一個最小完整應(yīng)用來源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])若需要啟動時建表/初始化數(shù)據(jù)可以像前文on_startup那樣使用config.get_engine()/config.get_session()顯式初始化。文檔特別強(qiáng)調(diào)示例中“先 drop 再 create”僅為了可重復(fù)運(yùn)行生產(chǎn)環(huán)境不應(yīng)這樣做——生產(chǎn)應(yīng)使用遷移工具管理表結(jié)構(gòu)。同步版本見 sqlalchemy_sync_plugin_example.py。運(yùn)行方式litestar run curl -X POST -H Content-Type: application/json \ -d {title: Your Todo Title, done: false} http://localhost:8000/五、SQLAlchemyInitPlugin依賴注入與會話生命周期如果你只需要數(shù)據(jù)庫基礎(chǔ)設(shè)施而不需要自動序列化使用SQLAlchemyInitPlugin。它為應(yīng)用提供四項能力通過依賴注入暴露 engine 與 session默認(rèn)鍵名為db_engine/db_session在應(yīng)用 state 中管理 engine 與 session factory配置一個before_send鉤子響應(yīng)發(fā)送前默認(rèn)關(guān)閉 session 并將其從連接 scope 中移除向 signature namespace 注入相關(guān)名稱輔助解析注解類型。最簡依賴注入示例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 與普通依賴一樣可以被其他依賴函數(shù)再次注入。5.1 重命名依賴鍵通過在配置對象上設(shè)置engine_dependency_key與session_dependency_key兩個屬性可以改變 engine 和 session 綁定的依賴名避免與業(yè)務(wù)依賴沖突。5.2 自定義 before_send 處理器默認(rèn)before_send處理器只負(fù)責(zé)關(guān)閉 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用于避免響應(yīng)序列化時觸發(fā)“detached instance”訪問異常配置對象本身還暴露若干控制插件行為的選項完整參數(shù)請查閱 Advanced-Alchemy 的 reference 文檔。僅用 Init Plugin 時Handler 內(nèi)部需要自行在 ORM 對象與可序列化結(jié)構(gòu)之間轉(zhuǎn)換SQLAlchemySerializationPlugin正是為消除這類樣板代碼而設(shè)計的。六、SQLAlchemySerializationPlugin自動 DTO 與字段標(biāo)記SQLAlchemySerializationPlugin不接收任何參數(shù)實例化后直接傳給應(yīng)用即可來源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()])工作原理應(yīng)用注冊階段凡是 handler 的data參數(shù)或返回注解為 SQLAlchemy 模型或模型集合、且未顯式指定 DTO 類的插件會為它們生成SQLAlchemyDTO。因此下面兩種寫法功能等價掛載SQLAlchemySerializationPlugin零聲明顯式定義 DTO 類見 sqlalchemy_async_serialization_dto.py。字段標(biāo)記控制出入數(shù)據(jù)既然插件只是“替 handler 定義 DTO”就可以用 DTO 的字段標(biāo)記機(jī)制mark控制哪些字段允許進(jìn)出。示例 sqlalchemy_async_serialization_plugin_marking_fields.py 中模型新增了一個super_secret_value屬性并在 handler 中賦值但該字段被標(biāo)記為 private 后序列化響應(yīng)中不會出現(xiàn)它——這是把敏感字段如內(nèi)部狀態(tài)、憑據(jù)擋在 API 邊界之外的標(biāo)準(zhǔn)做法。七、三個插件的選型關(guān)系與延伸閱讀插件職責(zé)適用場景SQLAlchemyPluginInit Serialization 全量能力最常見選擇快速接入SQLAlchemyInitPlugin引擎/會話的依賴注入、state 管理、before_send、namespace手動管理序列化只需數(shù)據(jù)庫基礎(chǔ)設(shè)施SQLAlchemySerializationPlugin為 ORM 模型注解自動生成 DTO已有會話管理只想免除 DTO 樣板代碼從源碼結(jié)構(gòu)看SQLAlchemyPlugin是對后兩者的組合封裝三者可以按需獨(dú)立掛載。倉庫中還提供了一組“同一應(yīng)用的四種裝配方式”對照示例便于橫向比較樣板代碼量的差異full_app_no_plugins.py、full_app_with_init_plugin.py、full_app_with_serialization_plugin.py 與 full_app_with_plugin.py。關(guān)鍵文檔與示例路徑一覽章節(jié)索引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需要強(qiáng)調(diào)的是適用前提以上所有示例均以 SQLiteaiosqlite/標(biāo)準(zhǔn)庫演示切換到 Postgres 等生產(chǎn)數(shù)據(jù)庫時只需更換connection_string的驅(qū)動前綴如postgresqlasyncpg://...模型、Repository 與插件代碼均可保持不變而表結(jié)構(gòu)變更請交由遷移工具而非create_all處理?！久赓M(fèi)下載鏈接】litestarLight, flexible and extensible ASGI framework | Built to scale項目地址: https://gitcode.com/GitHub_Trending/li/litestar創(chuàng)作聲明：本文部分內(nèi)容由AI輔助生成（AIGC），僅供參考