一旦握手完成，protocol changes from HTTP to WebSocket。如果您试图在WebSocket端点内引发HTTP异常，您会发现这是不可能的，或者返回HTTP响应（例如，return JSONResponse(...status_code=404)），您会得到一个内部服务器错误，即ASGI callable returned without sending handshake。

选项1

因此，如果您希望在协议升级之前有某种检查机制，则需要使用Middleware，如下所示：在中间件内部，您不能引发异常，但可以返回响应（即Response、JSONResponse、PlainTextResponse等），这实际上是FastAPI在后台处理异常的方式。作为参考，请看一下这个post，以及讨论here。

async def is_user_allowed(request: Request):
    # if conditions are not met, return False
    print(request['headers'])
    print(request.client)
    return False

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    if not await is_user_allowed(request):
        return JSONResponse(content={"message": "User not allowed"}, status_code=404)
    response = await call_next(request)
    return response

或者，如果愿意，可以让is_user_allowed()方法引发一个需要用try-except块捕获的自定义异常：

class UserException(Exception):
    def __init__(self, message):
        self.message = message
        super().__init__(message)

async def is_user_allowed(request: Request):
    # if conditions are not met, raise UserException
    raise UserException(message="User not allowed.")

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    try:
        await is_user_allowed(request)
    except UserException as e:
        return JSONResponse(content={"message": f'{e.message}'}, status_code=404)
    response = await call_next(request)
    return response

选项2

但是，如果需要使用websocket示例来执行此操作，则可以使用与上面相同的逻辑，但要改为在is_user_allowed()方法中传递websocket示例，并在WebSocket端点内捕获异常（受this启发）。

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    try:
        await is_user_allowed(ws)
        await handle_conn(ws)
    except UserException as e:
        await ws.send_text(e.message) # optionally send a message to the client before closing the connection
        await ws.close()

但是，在上面的代码中，你必须首先接受连接，这样如果出现异常，你就可以调用close()方法来终止连接。如果你愿意，你可以使用类似下面的代码。但是，except块中的return语句会引发一个内部服务器错误（即ASGI callable returned without sending handshake.），如前所述。

@app.websocket("/ws")
async def websocket_endpoint(ws: WebSocket):
    try:
        await is_user_allowed(ws)
    except UserException as e:
        return
    await ws.accept()
    await handle_conn(ws)