如何为Litestar应用全生命周期全局提供单次依赖?
如何为Litestar应用全生命周期全局提供单次依赖?
嘿,兄弟,看你现在的代码已经摸到门道了,稍微调整下就能实现“全生命周期只实例化一次”的全局依赖需求!
首先,你写的provide_jarpc_app异步生成器逻辑非常到位——用它封装RPC客户端的创建、连接,还通过finally块保证应用关闭时自动断开连接,资源的创建和清理逻辑都给包圆了。现在只需要在Litestar注册这个依赖时,明确给它打个“单例标记”就行。
给你补全后的完整实现代码,一看就懂:
import typing from litestar import Litestar, Provide from litestar.config.logging import LoggingConfig # 以下是你项目里的自定义导入,按需替换路径 from your_module import Database, BaseRPCClient, CabbageRPCClient from your_rpc_handlers import ParseEventRpcMethod, UpdateMarkingRpcMethod import config # 你的项目配置模块 async def provide_jarpc_app(database: Database) -> typing.AsyncGenerator[BaseRPCClient]: client = CabbageRPCClient( handlers=[ParseEventRpcMethod, UpdateMarkingRpcMethod], config=config, dependencies={'database': database}, ) await client.connect() try: yield client finally: await client.disconnect() def get_litestar_app() -> Litestar: logging_config = LoggingConfig( root={"level": "INFO", "handlers": ["console"]}, handlers={"console": {"class": "logging.StreamHandler"}}, ) # 核心操作就在这里!给依赖加上单例配置 return Litestar( route_handlers=[], # 替换成你实际的路由处理函数列表 dependencies={ "jarpc_client": Provide(provide_jarpc_app, singleton=True) }, logging_config=logging_config, )
我给你拆解几个关键细节:
singleton=True是核心:默认情况下Litestar的依赖会在每次被调用时重新实例化(比如请求级依赖),但加上这个参数后,Litestar会在应用启动瞬间就执行一次provide_jarpc_app,把生成的RPC客户端实例缓存起来,整个应用运行期间,不管路由、中间件还是其他依赖用到它,都是同一个实例。- 全局随便用:注册到应用的
dependencies字典后,任何地方只要通过类型提示或者依赖键(比如jarpc_client)就能拿到这个单例,举个路由里的使用例子:
from litestar import get @get("/call-rpc") async def test_rpc_call(jarpc_client: BaseRPCClient): # 直接用全局单例客户端发起RPC调用 rpc_result = await jarpc_client.call("your_rpc_method", params={"id": 123}) return {"rpc_result": rpc_result}
- 自动回收资源:当你关闭Litestar应用时,框架会自动触发异步生成器的
finally块,执行await client.disconnect(),不用你手动写代码去清理资源,省心得很。
这样调整后,你的RPC客户端就会在应用全生命周期里只初始化一次,全局复用,完全符合你的需求!
备注:内容来源于stack exchange,提问作者Pavel Shishmarev
相关产品推荐
相关产品推荐

