FastAPI 注入（fastapi_inject）

symphra_container.integrations.fastapi_inject(service_type)

创建 FastAPI 依赖注入函数.

用于在 FastAPI 路由中注入服务。返回的函数可以作为 Depends() 的参数。

参数：

名称	类型	描述	默认
service_type	type[T]	要注入的服务类型	必需

返回：

名称	类型	描述
Callable	Callable[[], T]	可用于 Depends() 的依赖函数

引发：

类型	描述
RuntimeError	如果容器未初始化
ServiceNotFoundError	如果服务未注册

示例

@app.get("/users") async def get_users( ... user_service: UserService = Depends(inject(UserService)) ... ): ... return await user_service.get_all()

源代码位于： src/symphra_container/integrations/fastapi.py

def inject(service_type: type[T]) -> Callable[[], T]:
    """创建 FastAPI 依赖注入函数.

    用于在 FastAPI 路由中注入服务。返回的函数可以作为 Depends() 的参数。

    Args:
        service_type: 要注入的服务类型

    Returns:
        Callable: 可用于 Depends() 的依赖函数

    Raises:
        RuntimeError: 如果容器未初始化
        ServiceNotFoundError: 如果服务未注册

    示例:
        >>> @app.get("/users")
        >>> async def get_users(
        ...     user_service: UserService = Depends(inject(UserService))
        ... ):
        ...     return await user_service.get_all()
    """

    def dependency() -> T:
        container = get_container()
        # 检查是否是异步服务
        registration = container._registrations.get(service_type)
        if registration and registration.is_async:
            # 异步服务需要用 resolve_async, 但这里是同步上下文
            # FastAPI 会自动处理异步依赖，所以这里直接返回 coroutine
            import asyncio

            coro = container.resolve_async(service_type)
            # 如果当前已经在 async 上下文中，直接返回 coroutine
            # FastAPI 会自动 await 它
            try:
                asyncio.get_running_loop()
                # 在 async 上下文中，返回 coroutine 让 FastAPI await
                return coro  # type: ignore
            except RuntimeError:
                # 不在 async 上下文中，尝试同步解析
                return container.resolve(service_type)
        else:
            # 同步服务直接解析
            return container.resolve(service_type)

    return dependency