跳到内容

异常 - HTTPExceptionWebSocketException

这些是您可以抛出的异常,用于向客户端显示错误。

当您抛出异常时,就像正常的 Python 一样,其余的执行将被中止。这样,您可以从代码的任何位置抛出这些异常来中止请求并向客户端显示错误。

您可以使用

  • HTTPException
  • WebSocketException

这些异常可以直接从 fastapi 导入。

from fastapi import HTTPException, WebSocketException

fastapi.HTTPException

HTTPException(status_code, detail=None, headers=None)

Bases: HTTPException

您可以在自己的代码中抛出的 HTTP 异常,用于向客户端显示错误。

这适用于客户端错误、无效身份验证、无效数据等。不适用于代码中的服务器错误。

FastAPI 错误处理文档 中了解更多关于它的信息。

示例

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}


@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}
参数 描述
status_code

发送给客户端的 HTTP 状态码。

类型: int

detail

要在 JSON 响应的 detail 键中发送给客户端的任何数据。

类型: Any 默认值: None

headers

要在响应中发送给客户端的任何标头。

类型: Optional[dict[str, str]] 默认值: None

fastapi/exceptions.py 中的源代码
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def __init__(
    self,
    status_code: Annotated[
        int,
        Doc(
            """
            HTTP status code to send to the client.
            """
        ),
    ],
    detail: Annotated[
        Any,
        Doc(
            """
            Any data to be sent to the client in the `detail` key of the JSON
            response.
            """
        ),
    ] = None,
    headers: Annotated[
        Optional[dict[str, str]],
        Doc(
            """
            Any headers to send to the client in the response.
            """
        ),
    ] = None,
) -> None:
    super().__init__(status_code=status_code, detail=detail, headers=headers)

status_code instance-attribute

status_code = status_code

detail instance-attribute

detail = detail

headers instance-attribute

headers = headers

fastapi.WebSocketException

WebSocketException(code, reason=None)

Bases: WebSocketException

您可以在自己的代码中抛出的 WebSocket 异常,用于向客户端显示错误。

这适用于客户端错误、无效身份验证、无效数据等。不适用于代码中的服务器错误。

FastAPI 文档的“WebSockets”中了解更多。

示例

from typing import Annotated

from fastapi import (
    Cookie,
    FastAPI,
    WebSocket,
    WebSocketException,
    status,
)

app = FastAPI()

@app.websocket("/items/{item_id}/ws")
async def websocket_endpoint(
    *,
    websocket: WebSocket,
    session: Annotated[str | None, Cookie()] = None,
    item_id: str,
):
    if session is None:
        raise WebSocketException(code=status.WS_1008_POLICY_VIOLATION)
    await websocket.accept()
    while True:
        data = await websocket.receive_text()
        await websocket.send_text(f"Session cookie is: {session}")
        await websocket.send_text(f"Message text was: {data}, for item ID: {item_id}")
参数 描述
code

规范中定义的有效代码 中的关闭代码。

类型: int

reason

关闭 WebSocket 连接的原因。

它是 UTF-8 编码的数据。原因的解释取决于应用程序,WebSocket 规范并未指定。

它可以包含人类可读的文本,或可由客户端代码解释的文本等。

类型: 联合类型[字符串, None] 默认值: None

fastapi/exceptions.py 中的源代码
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
def __init__(
    self,
    code: Annotated[
        int,
        Doc(
            """
            A closing code from the
            [valid codes defined in the specification](https://datatracker.ietf.org/doc/html/rfc6455#section-7.4.1).
            """
        ),
    ],
    reason: Annotated[
        Union[str, None],
        Doc(
            """
            The reason to close the WebSocket connection.

            It is UTF-8-encoded data. The interpretation of the reason is up to the
            application, it is not specified by the WebSocket specification.

            It could contain text that could be human-readable or interpretable
            by the client code, etc.
            """
        ),
    ] = None,
) -> None:
    super().__init__(code=code, reason=reason)

code instance-attribute

code = code

reason instance-attribute

reason = reason or ''