我正在尝试使用 FastAPI 创建一个带有身份验证的简单 WebSocket 应用程序,但添加身份验证后我遇到了问题,
/{id}
请求通过,但WebSocket连接/ws/{client_id}
由于以下错误而失败
ERROR: Exception in ASGI application
Traceback (most recent call last):
File "D:\Installations\envs\chat-app\lib\site-packages\uvicorn\protocols\websockets\websockets_impl.py", line 240, in run_asgi
result = await self.app(self.scope, self.asgi_receive, self.asgi_send)
File "D:\Installations\envs\chat-app\lib\site-packages\uvicorn\middleware\proxy_headers.py", line 69, in __call__
return await self.app(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\applications.py", line 1054, in __call__
await super().__call__(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\applications.py", line 123, in __call__
await self.middleware_stack(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\middleware\errors.py", line 151, in __call__
await self.app(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\middleware\exceptions.py", line 65, in __call__
await wrap_app_handling_exceptions(self.app, conn)(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
raise exc
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
await app(scope, receive, sender)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 756, in __call__
await self.middleware_stack(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 776, in app
await route.handle(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 373, in handle
await self.app(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 96, in app
await wrap_app_handling_exceptions(app, session)(scope, receive, send)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 64, in wrapped_app
raise exc
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\_exception_handler.py", line 53, in wrapped_app
await app(scope, receive, sender)
File "D:\Installations\envs\chat-app\lib\site-packages\starlette\routing.py", line 94, in app
await func(session)
File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\routing.py", line 338, in app
solved_result = await solve_dependencies(
File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\dependencies\utils.py", line 572, in solve_dependencies
solved_result = await solve_dependencies(
File "D:\Installations\envs\chat-app\lib\site-packages\fastapi\dependencies\utils.py", line 600, in solve_dependencies
solved = await call(**sub_values)
TypeError: HTTPBasic.__call__() missing 1 required positional argument: 'request'
INFO: connection open
INFO: connection closed
代码:
app = FastAPI()
security = HTTPBasic()
templates = Jinja2Templates(directory="templates")
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str, websocket: WebSocket) :
for connection in self.active_connections:
if (connection == websocket):
continue
await connection.send_text(message)
connectionmanager = ConnectionManager()
def get_current_username(
credentials: Annotated[HTTPBasicCredentials, Depends(security)],
):
is_correct_username, is_correct_password= validate_user(credentials.username.encode("utf8"),credentials.password.encode("utf8"))
if not (is_correct_username and is_correct_password):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Basic"},
)
return credentials.username
@app.get("/{id}", response_class=HTMLResponse)
def read_user(id: int, username: Annotated[str, Depends(get_current_username)], request: Request):
# Render the HTML template
return templates.TemplateResponse("index.html", {"request": request})
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(
websocket: WebSocket, client_id: int, username: Annotated[str, Depends(get_current_username)], request: Request):
# accept connections
await connectionmanager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await connectionmanager.send_personal_message(f"You : {data}", websocket)
await connectionmanager.broadcast(f"Client #{client_id}: {data}", websocket)
except WebSocketDisconnect:
connectionmanager.disconnect(websocket)
await connectionmanager.broadcast(f"Client #{client_id} left the chat")
if __name__ == "__main__":
uvicorn.run(app, host='localhost', port=8000)
在添加身份验证机制之前,它工作正常。
即使我添加了请求参数,我仍然收到此错误。
你找到解决办法了吗?