FastAPI中如何向自定义WebSocket处理类的方法注入依赖？

FastAPI内置的依赖解析工具支持WebSocket场景，不需要引入额外第三方容器即可实现你要的效果：

import abc
from fastapi import WebSocket, Depends
from fastapi.dependencies.utils import solve_dependencies

class Handler:
    async def __call__(self, websocket: WebSocket):
        await websocket.accept()
        # 解析handle_websocket方法的所有依赖
        solved = await solve_dependencies(
            request=websocket,
            dependant=self.handle_websocket,
            dependency_overrides_provider=None
        )
        values, errors, *_ = solved
        if errors:
            await websocket.close(code=1008)
            return
        await self.handle_websocket(**values)

    @abc.abstractmethod
    async def handle_websocket(self, websocket: WebSocket):
        pass

子类使用示例：

# 示例依赖
async def get_db_session():
    # 模拟数据库会话生成逻辑
    session = AsyncSession()
    try:
        yield session
    finally:
        await session.close()

# 业务实现类
class ChatHandler(Handler):
    async def handle_websocket(self, websocket: WebSocket, db = Depends(get_db_session)):
        # db参数会被自动注入，无需手动传递
        while True:
            msg = await websocket.receive_text()
            await websocket.send_text(f"已接收消息：{msg}")

如果你项目中已经使用了第三方DI容器，直接调用容器提供的执行方法，传入手动参数websocket即可，其余参数容器会自动完成注入：

# DI容器定义示例
from dependency_injector import containers, providers
from sqlalchemy.ext.asyncio import AsyncSession

class AppContainer(containers.DeclarativeContainer):
    db_session = providers.Factory(AsyncSession)

# 全局容器获取方法
def get_dependency_injection_container() -> AppContainer:
    return AppContainer()

# Handler类实现
class Handler:
    async def __call__(self, websocket: WebSocket):
        await websocket.accept()
        container = get_dependency_injection_container()
        # 仅传手动参数websocket，其余依赖由容器自动填充
        await container.call(self.handle_websocket, websocket=websocket)

    @abc.abstractmethod
    async def handle_websocket(self, websocket: WebSocket, db_session: AsyncSession):
        pass

• 你提供的初始代码中类实例方法缺失了self参数，需要先修正，否则会出现调用参数不匹配的报错
• 可以在__call__方法中增加异常捕获逻辑，依赖注入或业务执行出错时主动关闭WebSocket连接，避免连接泄漏

内容的提问来源于stack exchange，提问作者Ruberoid