Litestar 依赖注入实战分层声明、Provide 包装器与 yield 清理机制全解析【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar在 LitestarLight, flexible and extensible ASGI framework中依赖注入Dependency InjectionDI是把数据库会话、外部服务客户端、配置对象等横切逻辑从路由处理函数中剥离出来的核心手段。本篇基于官方文档 docs/usage/dependency-injection.rst 展开带你掌握如何在 App、Router、Controller、Route Handler 四个层级声明依赖如何用Provide包装可调用对象并控制use_cache、sync_to_thread如何用 yield 依赖实现“请求结束后自动清理连接”的上下文管理以及 2.24 版本引入的NamedDependency显式标记机制如何避免名称推断带来的隐式行为。读完并结合 litestar/di.py 与 litestar/_kwargs/dependencies.py 的源码你将能够独立完成生产级应用的依赖装配与请求级生命周期管理。1. 四个层级上声明依赖scope 决定可见性Litestar 的依赖注入系统允许在应用的每一层声明依赖。官方示例如下它同时展示了四种声明位置from litestar import Controller, Router, Litestar, get from litestar.di import NamedDependency, Provide async def bool_fn() - bool: ... async def dict_fn() - dict: ... async def list_fn() - list: ... async def int_fn() - int: ... class MyController(Controller): path /controller # 声明在 Controller 层 dependencies {controller_dependency: Provide(list_fn)} # 声明在 Route Handler 层 get(path/handler, dependencies{local_dependency: Provide(int_fn)}) def my_route_handler( self, app_dependency: NamedDependency[bool], router_dependency: NamedDependency[dict], controller_dependency: NamedDependency[list], local_dependency: NamedDependency[int], ) - None: ... # 声明在 Router 层 my_router Router( path/router, dependencies{router_dependency: Provide(dict_fn)}, route_handlers[MyController], ) # 声明在 App 层 app Litestar( route_handlers[my_router], dependencies{app_dependency: Provide(bool_fn)} )这段代码对应 docs/usage/dependency-injection.rst 开篇的主例其核心规则是依赖被隔离在声明它的上下文scope内local_dependency只能在声明它的那个路由处理函数内访问controller_dependency只对该 Controller 上的所有路由处理函数可用router_dependency只对注册在该 Router 下的处理函数可用只有app_dependency对所有路由处理函数可见。依赖的注入值最终通过处理函数签名中带有NamedDependency标记的参数接收。文档同时提醒Litestar 需要在运行时获取被注入的类型信息这与 linter 推荐的TYPE_CHECKING条件导入写法存在冲突相关处理参见文档中的 Signature namespace 主题示例。版本注意自 2.24 起依赖不再仅凭参数名与依赖 key 匹配就自动注入。依赖旧版“按名推断”行为的代码会触发LitestarDeprecationWarning并在 Litestar 3.0 中彻底失效——必须改用NamedDependency显式标记见第 6 节。2. 依赖注入的五项前置条件官方文档列出了依赖注入生效的全部前提逐条对照可以避免绝大多数“依赖没被注入”的困惑依赖必须是可调用对象callable依赖只能接收 kwargs 和一个self参数不能接收位置参数接收依赖的参数必须标记为NamedDependency且参数名必须与依赖 key 一致依赖必须用Provide类声明包装后放入各层的dependencies字典依赖必须处于处理函数的 scope 之内即第 1 节描述的层级可见性。第 2 条的原因在“依赖的 kwargs”一节有更深解释依赖函数与路由处理函数使用同一套签名解析机制见第 4 节源码印证因此它同样可以声明查询参数、路径参数等。3. 同步与异步可调用对象sync_to_thread 的取舍文档中内嵌了一条重要通告docs/admonitions/sync-to-thread-info.rst对同步/异步依赖同样适用同步、异步可调用对象都受支持执行阻塞 I/O 或重计算的同步函数可能阻塞事件循环线程进而卡住整个应用为此Provide提供sync_to_thread参数设为True时函数会在线程池中运行若同步函数确认是非阻塞的显式设sync_to_threadFalse可告知 Litestar 无需线程切换不设置sync_to_thread而传入同步函数时Litestar 会发出告警。从源码 litestar/di.py 可以印证这套逻辑Provide.__init__先判定has_sync_callable若sync_to_threadTrue且依赖是同步可调用对象则用ensure_async_callable包装成异步形式对同步可调用对象未显式指定sync_to_thread时调用warn_implicit_sync_to_thread发出告警。这与单测 tests/unit/test_di.py 中test_sync_callable_without_sync_to_thread_warns断言的“discouraged since synchronous callables”告警行为完全吻合。4. yield 依赖在 handler 返回后执行清理步骤除了普通可调用对象依赖还可以是异步生成器函数。yield之前的代码负责建立资源打开连接、创建会话yield之后的代码作为清理步骤在 handler 返回后执行。技术细节来自文档通告框清理阶段发生在 handler 函数返回之后但 HTTP 响应发送之前。4.1 基础示例官方示例 docs/examples/dependency_injection/dependency_yield_simple.py# dependencies.py from collections.abc import Generator from litestar import Litestar, get from litestar.di import Provide CONNECTION {open: False} def generator_function() - Generator[dict[str, bool], None, None]: Set connection to open and close it after the handler returns. CONNECTION[open] True yield CONNECTION CONNECTION[open] False get(/, dependencies{conn: Provide(generator_function)}) def index(conn: dict[str, bool]) - dict[str, bool]: Return the current connection state. return conn app Litestar(route_handlers[index])运行下面的测试代码可以看到 handler 返回后CONNECTION已被重置from litestar.testing import TestClient from dependencies import app, CONNECTION with TestClient(appapp) as client: print(client.get(/).json()) # {open: True} print(CONNECTION) # {open: False}从源码看这个机制在 litestar/_kwargs/dependencies.py 中实现resolve_dependency调用Provide后若has_sync_generator_dependency为真则把生成器加入DependencyCleanupGroup并执行next(value)取得yield值异步生成器则用await anext(value)。请求结束时由清理组统一推进生成器到yield之后的代码。4.2 异常处理在依赖内感知 handler 异常如果 handler 抛出异常异常会被注入throw到生成器内第一次yield的位置。这让依赖可以根据异常决定行为例如出错时回滚数据库会话、正常时提交。官方示例 docs/examples/dependency_injection/dependency_yield_exceptions.py# dependencies.py from collections.abc import Generator from litestar import Litestar, get from litestar.di import Provide STATE {result: None, connection: closed} def generator_function() - Generator[str, None, None]: Set the connection state to open and close it after the handler returns. If an error occurs, set result to error, else set it to OK. try: STATE[connection] open yield hello STATE[result] OK except ValueError: STATE[result] error finally: STATE[connection] closed get(/{name:str}, dependencies{message: Provide(generator_function)}) def index(name: str, message: str) - dict[str, str]: If name is John, return a message, otherwise raise an error. if name John: return {name: message} raise ValueError() app Litestar(route_handlers[index])验证输出from litestar.testing import TestClient from dependencies import STATE, app with TestClient(appapp) as client: response client.get(/John) print(response.json()) # {John: hello} print(STATE) # {result: OK, connection: closed} response client.get(/Peter) print(response.status_code) # 500 print(STATE) # {result: error, connection: closed}两条最佳实践来自文档原文始终用try/finally包裹yield即使你并不想处理异常也能保证异常发生时清理代码依然执行def generator_dependency(): try: yield finally: ... # cleanup code不要在依赖内重新抛出re-raise捕获的异常——捕获到的异常仍会经由框架的常规机制处理无需显式 re-raise。关于清理阶段自身抛出异常的语义文档还给出关键说明清理步骤中抛出的异常会打包进ExceptionGroup重新抛出Python 3.11 使用 exceptiongroup 包且发生在所有依赖完成清理之后——因此某一个依赖清理失败不会影响其他依赖的清理。从源码印证litestar/_kwargs/cleanup.py 的DependencyCleanupGroup._throw会依次对每个生成器执行gen.throw(exc)/await gen.athrow(exc)收集清理过程中新产生的异常最后以ExceptionGroup(Exceptions occurred during cleanup of dependencies, exceptions)形式抛出而正常路径的_cleanupL77-L96会在多生成器场景下用anyio.TaskGroup逆序并发推进各生成器完成清理。4.3 缓存与生成器互斥从 litestar/di.py 可以看到一个硬约束Provide(use_cacheTrue)与生成器依赖不能同时使用否则会直接抛出ImproperlyConfiguredException(Cannot cache generator dependency, consider using Lifespan Context instead.)。原因是生成器依赖有状态yield 前后各执行一次缓存其返回值没有意义应用级、跨请求的缓存需求应转向 Lifespan 上下文管理。5. 依赖的 kwargs依赖函数也能“被注入”文档明确指出依赖函数与路由处理函数用同一机制解析签名因此依赖可以注入与路由处理函数相同的保留关键字参数handler 的“reserved keyword arguments”如request、state等。官方示例from litestar import Controller, patch from litestar.di import NamedDependency, Provide from pydantic import BaseModel, UUID4 class User(BaseModel): id: UUID4 name: str async def retrieve_db_user(user_id: UUID4) - User: ... class UserController(Controller): path /user dependencies {user: Provide(retrieve_db_user)} patch(path/{user_id:uuid}) async def get_user(self, user: NamedDependency[User]) - User: ...示例中User模型由辅助函数retrieve_db_user从数据库取出它通过user_idkwarg由路径参数/{user_id:uuid}解析而来获取对应实例。UserController把retrieve_db_user以 keyuser注册进dependencies字典于是get_user的user参数就自动拿到查好的User实例——依赖内部完全不必接触路由细节。从源码印证litestar/_kwargs/dependencies.py 的resolve_dependency先取dependency.provide.signature_model若该签名模型有字段就调用parse_values_from_connection_kwargs从当前连接Request/WebSocket和已有 kwargs 中解析依赖自己的参数而Provide.finalizelitestar/di.py在应用装配期用SignatureModel.create(dependency_name_setdependency_keys, ...)完成这一解析模型的构建——dependency_name_set正是当前 scope 内的依赖 key 集合这保证了依赖函数中名为依赖 key 的参数会先注入子依赖而不是被误当查询参数处理。此外依赖图是分批次解析的create_dependency_batches 递归展开DependencyContainer的子依赖反复挑选“子依赖已全部解析”的节点组成一个批次从而保证“依赖的依赖”按拓扑顺序先后求值。6.NamedDependency标记2.24 后的显式契约2.24 之前的版本依赖名称推断只要参数名恰好等于某个 scope 内依赖的 key就会自动注入无需任何标记。这种隐式行为容易因重名而误注入因此被废弃。现在的规则是应接收依赖的参数必须标记为NamedDependency[T]泛型参数名与 scope 内的依赖 key 匹配NamedDependency包裹的类型用于校验注入进来的值。from litestar import get from litestar.di import NamedDependency, Provide async def my_dependency() - int: ... get(/, dependencies{my_dep: my_dependency}) def handler(my_dep: NamedDependency[int]) - None: ...而被废弃的写法——仅靠参数名匹配注入# 已废弃my_dep 之所以被注入仅仅因为参数名恰好等于依赖 key。 # 请改用 NamedDependency[int] 标记。 get(/, dependencies{my_dep: Provide(my_dependency)}) def handler(my_dep: int) - None: ...从源码看NamedDependency的实现非常轻量——litestar/di.py 中它只是Annotated[T, Dependency(kindnamed)]即通过Annotated元数据携带一个kindnamed的Dependency标记对象源码注释指出 3.0 将引入基于类型的 type kind。框架在签名解析阶段识别该标记将其与 scope 内的依赖 key 关联起来。6.1 带默认值的依赖自动排除出 OpenAPI 文档如果处理函数或Provide函数声明的依赖参数带有默认值且该路由并未真正提供这个依赖则应回退使用默认值。只要把参数声明为依赖标记NamedDependencyLitestar 就明白它是依赖而非查询参数会将其从 OpenAPI 文档中排除# docs/examples/dependency_injection/dependency_with_dependency_fn_and_default.py from typing import Any from litestar import Litestar, get from litestar.di import NamedDependency get(/) async def hello_world(optional_dependency: NamedDependency[int] 3) - dict[str, Any]: Notice we havent provided the dependency to the route. This is OK, because of the default value, and now the parameter is excluded from the docs. return {hello: optional_dependency} app Litestar(route_handlers[hello_world])6.2 未提供且无默认值启动期即报错早期失败同一枚硬币的另一面依赖未提供且没有默认值时若参数没有NamedDependency标记它会被当作普通未标记 handler kwarg——触发一条关于“缺少显式参数标记”的LitestarDeprecationWarning并且路由要到请求时刻才因为找不到匹配的查询参数而失败。加上NamedDependency后契约显式化Litestar 在应用启动时就拒绝启动# docs/examples/dependency_injection/dependency_non_optional_not_provided.py from typing import Any from litestar import Litestar, get from litestar.di import NamedDependency get(/) async def hello_world(non_optional_dependency: NamedDependency[int]) - dict[str, Any]: Notice we havent provided the dependency to the route. This is not great, however by explicitly marking dependencies, Litestar wont let the app start. return {hello: non_optional_dependency} app Litestar(route_handlers[hello_world]) # ImproperlyConfiguredException: 500: Explicit dependency non_optional_dependency for hello_world has no default # value, or provided dependency.这种“fail fast”是显式标记机制最大的工程价值把配置错误从线上请求暴露提前到启动时暴露。7. 依赖覆盖Override低层级覆盖高层级因为每一层都用字符串 key 的字典声明依赖覆盖依赖非常简单——低层级用同一个 key 重新声明即可低层级依赖会覆盖高层级依赖from litestar import Controller, get from litestar.di import NamedDependency, Provide def bool_fn() - bool: ... def dict_fn() - dict: ... class MyController(Controller): path /controller # Controller 层 dependencies {some_dependency: Provide(dict_fn)} # Route Handler 层同 key 覆盖 get(path/handler, dependencies{some_dependency: Provide(bool_fn)}) def my_route_handler( self, some_dependency: NamedDependency[bool], ) - None: ...注意覆盖后注入类型随之变化dict→boolNamedDependency的类型参数也需要同步更新。依赖嵌套下一节时覆盖规则同样适用。8.Provide类详解Provide是依赖注入的包装器任何要注入的可调用对象都必须用Provide包装。依赖可以是同步/异步函数、方法或实现了__call__的类实例也可以直接是类。from random import randint from litestar import get from litestar.di import NamedDependency, Provide def my_dependency() - int: return randint(1, 10) get( /some-path, dependencies{ my_dep: Provide( my_dependency, ) }, ) def my_handler(my_dep: NamedDependency[int]) - None: ...Provide的完整构造函数签名摘自 litestar/di.py参数类型默认值说明dependencyAnyCallable \| type[Any]必填要调用的可调用对象或要实例化的类非可调用对象会抛ImproperlyConfiguredExceptionuse_cacheboolFalse是否缓存依赖的返回值见下方说明sync_to_threadbool \| NoneNone是否在事件循环线程外运行同步依赖None时对同步函数发出告警关于use_cache文档有明确告诫当Provide.use_cache为True时函数返回值在首次调用后被记忆化memoized后续直接复用实现上没有kwargs 比较、LRU 等复杂逻辑因此要谨慎使用。同时注意即使use_cacheFalse依赖在单个请求内也只会被调用一次。从源码 litestar/di.py 可以看到__call__的缓存逻辑就是简单的self.value is not Empty判断tests/unit/test_di.py 的test_provide_cached验证了连续三次调用返回同一缓存值。9. 依赖中注入依赖依赖函数本身也可以注入其他依赖用法与处理函数完全一致。官方示例from litestar import Litestar, get from litestar.di import NamedDependency, Provide from random import randint async def first_dependency() - int: return randint(1, 10) async def second_dependency(injected_integer: NamedDependency[int]) - bool: return injected_integer % 2 0 get(/true-or-false) def true_or_false_handler(injected_bool: NamedDependency[bool]) - str: return its true! if injected_bool else nope, its false... app Litestar( route_handlers[true_or_false_handler], dependencies{ injected_integer: first_dependency, injected_bool: second_dependency, }, )这里的调用链是injected_integer先求值 → 结果作为 kwarg 注入second_dependency→ 其布尔结果再注入true_or_false_handler。文档同时提醒依赖覆盖规则在此同样适用更高层级若声明了同 key 依赖会被优先采用。10. 跳过依赖值校验默认情况下Litestar 会对注入的依赖值进行类型校验。比如 handler 声明NamedDependency[int]而依赖实际返回str会得到内部服务器错误# docs/examples/dependency_injection/dependency_validation_error.py from typing import Any from litestar import Litestar, get from litestar.di import NamedDependency, Provide async def provide_str() - str: Returns a string. return whoops get(/, dependencies{injected: Provide(provide_str)}, sync_to_threadFalse) def hello_world(injected: NamedDependency[int]) - dict[str, Any]: Handler expects an int, but weve provided a str. return {hello: injected} app Litestar(route_handlers[hello_world])可以用SkipValidation标记绕过这一校验注意绕过校验时参数不再是NamedDependency名称推断场景下仍依赖 key 匹配在显式标记场景下按你的注入契约处理# docs/examples/dependency_injection/dependency_skip_validation.py from typing import Any from litestar import Litestar, get from litestar.params import SkipValidation async def provide_str() - str: Returns a string. return whoops get(/, dependencies{injected: provide_str}) async def hello_world(injected: SkipValidation[int]) - dict[str, Any]: Handler expects an int, but weve provided a str. return {hello: injected} app Litestar(route_handlers[hello_world])SkipValidation适用于你确定依赖返回值形态、想省去校验开销或依赖返回类型在静态类型层面无法完整表达的场景。小结与延伸阅读本文以 docs/usage/dependency-injection.rst 为主线完整覆盖了 Litestar 依赖注入的全部要点层级与 scopeApp → Router → Controller → Route Handler 逐层声明低层级同 key 覆盖高层级契约显式化2.24 起必须用NamedDependency[T]标记注入参数未提供且无默认值的依赖在启动期即报ImproperlyConfiguredExceptionProvide三参数dependency可调用/类、use_cache请求外记忆化无 kwargs 比较、sync_to_thread同步依赖的线程化策略yield 依赖yield后代码作为清理步骤handler 返回后、响应发送前执行异常会被 throw 进生成器清理异常最终聚合为ExceptionGroup依赖的 kwargs 与嵌套依赖依赖函数与 handler 共用签名解析机制依赖链按拓扑批次求值。想继续深入可以从以下仓库路径入手实现层 litestar/di.py、解析与批次计算 litestar/_kwargs/dependencies.py、清理组 litestar/_kwargs/cleanup.py行为验证可参考 tests/unit/test_di.py 与 e2e 测试 tests/e2e/test_dependency_injection同步/异步取舍的完整论述见 docs/topics/sync-vs-async.rst显式声明主题见 docs/topics/explicit_declarations.rst。【免费下载链接】litestarLight, flexible and extensible ASGI framework | Built to scale项目地址: https://gitcode.com/GitHub_Trending/li/litestar创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考