如何在Litestar控制器中正确使用dependency-injector?
问题:Litestar控制器集成dependency-injector依赖注入失败
问题背景
使用Litestar开发项目原型时,希望通过dependency-injector框架实现控制器的依赖注入,但无法像FastAPI那样通过Depends直接集成,运行时出现类型不支持的错误。
项目结构
lite/ ├── __init__.py ├── __main__.py ├── controllers.py ├── di.py ├── logging.py ├── main.py ├── services.py
现有代码片段
di.py
from dependency_injector import containers, providers from .services import Service class Container(containers.DeclarativeContainer): service = providers.Factory(Service)
main.py
import uvicorn from .di import Container from .logging import logger if __name__ == "__main__": try: container = Container() container.wire(modules=[__name__, "lite", "lite.controllers"]) config = uvicorn.Config(app="lite.main:app", port=3000) server = uvicorn.Server(config) server.run() except Exception as e: logger.exception(e)
main.py
from litestar import Litestar from .controllers import ExampleController from litestar.logging import LoggingConfig app = Litestar( route_handlers=[ExampleController], logging_config=LoggingConfig( root={"level": "INFO", "handlers": ["queue_listener"]}, formatters={ "standard": { "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s" } }, log_exceptions="always", ), )
失败的控制器实现
from typing import Annotated from litestar import Controller, get from litestar.di import Provide as Depends from litestar.params import Dependency from .di import Container from dependency_injector.wiring import Provide, inject from .services import Service class ExampleController(Controller): path = "/test" @get("/") @inject async def bar(self, service: Service = Depends(Provide[Container.service])) -> None: service.foo() # 仅打印"foo"
错误信息
Traceback (most recent call last): File "/workspaces/LiteTest/.venv/lib/python3.10/site-packages/litestar/_signature/model.py", line 101, in _deserializer return default_deserializer(target_type, value) File "/workspaces/LiteTest/.venv/lib/python3.10/site-packages/litestar/serialization/msgspec_hooks.py", line 139, in default_deserializer raise TypeError(f"Unsupported type: {type(value)!r}") TypeError: Unsupported type: <class 'dependency_injector.wiring.Provide'> The above exception was the direct cause of the following exception: Traceback (most recent call last): File "/workspaces/LiteTest/.venv/lib/python3.10/site-packages/litestar/_signature/model.py", line 203, in parse_values_from_connection_kwargs return convert(kwargs, cls, strict=False, dec_hook=deserializer, str_keys=True).to_dict() msgspec.ValidationError: Unsupported type: <class 'dependency_injector.wiring.Provide'> - at `$.service`
已尝试的方案
- 直接在方法参数中使用
Depends(Provide[...])注入 - 尝试重写控制器构造函数,但Controller基类不可修改
- 查阅Litestar和dependency-injector官方文档,无相关集成指导
- 找到一种不够优雅的变通方法:
from litestar import Controller, get from litestar.di import Provide as Depends from .di import Container, Provide, inject from .services import Service @inject def _get_service(service: Service = Provide[Container.service]) -> Service: return service def get_service() -> Service: return _get_service() class ExampleController(Controller): path = "/test" @get("/", dependencies={"service": Depends(get_service)}) async def bar(self, service: Service) -> None: service.foo()
解决方案:正确集成dependency-injector与Litestar
问题根源是Litestar的参数解析系统无法识别dependency-injector的Provide类型,且同时混用了两套依赖注入机制导致冲突。以下是两种优雅的解决方式:
方案1:纯用dependency-injector的注入能力
去掉Litestar的Depends,仅使用dependency-injector的@inject和Provide,确保装饰器顺序正确(@inject靠近函数定义,让它先于Litestar路由装饰器执行):
from litestar import Controller, get from dependency_injector.wiring import inject, Provide from .di import Container from .services import Service class ExampleController(Controller): path = "/test" @get("/") @inject async def bar(self, service: Service = Provide[Container.service]) -> None: service.foo()
注意:确保__main__.py中container.wire(modules=["lite.controllers"])已正确包含控制器模块,让dependency-injector能扫描到需要注入的参数。
方案2:对接Litestar的依赖注入系统
将dependency-injector容器注册为Litestar的应用级依赖,通过Litestar的依赖提供者获取服务实例:
- 修改
di.py,添加全局容器实例:
from dependency_injector import containers, providers from .services import Service class Container(containers.DeclarativeContainer): service = providers.Factory(Service) # 创建全局容器实例 container = Container()
- 修改
main.py,初始化容器并注册为应用依赖:
from litestar import Litestar from .controllers import ExampleController from litestar.logging import LoggingConfig from .di import container # 初始化容器资源 container.init_resources() app = Litestar( route_handlers=[ExampleController], logging_config=LoggingConfig( root={"level": "INFO", "handlers": ["queue_listener"]}, formatters={ "standard": { "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s" } }, log_exceptions="always", ), # 注册容器为应用级依赖 dependencies={"container": container}, )
- 修改控制器,通过Litestar依赖系统获取服务:
from litestar import Controller, get from litestar.di import Provide from .di import Container from .services import Service # 定义依赖提供者函数 def get_service(container: Container) -> Service: return container.service() class ExampleController(Controller): path = "/test" @get("/", dependencies={"service": Provide(get_service)}) async def bar(self, service: Service) -> None: service.foo()
内容的提问来源于stack exchange,提问作者Arthur Brenno Ribeiro Coelho
相关产品推荐
相关产品推荐

