路由处理器¶

路由处理器是 Litestar 的核心。它们是通过使用从 Litestar 导出的处理器装饰器来装饰函数或类方法来构造的。

例如：

使用 @get() 装饰器装饰函数来定义路由处理器¶

from litestar import get


@get("/")
def greet() -> str:
   return "hello world"

在上面的示例中，装饰器包含了为路径 "/" 和 HTTP 动词 GET 的组合定义端点操作所需的所有信息。在这种情况下，它将是一个带有 Content-Type 头为 text/plain 的 HTTP 响应。

同步和异步可调用对象

支持同步和异步可调用对象。其中一个重要方面是，使用执行阻塞操作（例如 I/O 或计算密集型任务）的同步函数可能会阻塞运行事件循环的主线程，进而阻塞整个应用程序。

为了缓解这种情况，可以将 sync_to_thread 参数设置为 True，这将导致该函数在线程池中运行。

如果同步函数是非阻塞的，将 sync_to_thread 设置为 False 将告知 Litestar 用户确信其行为，并且该函数可以被视为非阻塞的。

如果传递了同步函数而未设置显式的 sync_to_thread 值，则会引发警告。

参见

同步与异步

声明路径¶

所有路由处理器装饰器都接受一个可选的路径参数。这个参数可以使用 path 参数作为关键字参数声明：

通过将路径作为关键字参数传递来定义路由处理器¶

from litestar import get


@get(path="/some-path")
async def my_route_handler() -> None: ...

它也可以作为参数传递而不使用关键字：

定义路由处理器但不使用关键字参数¶

from litestar import get


@get("/some-path")
async def my_route_handler() -> None: ...

这个参数的值可以是字符串路径（如上面的示例），也可以是 list 类型的 字符串 路径：

定义具有多个路径的路由处理器¶

from litestar import get


@get(["/some-path", "/some-other-path"])
async def my_route_handler() -> None: ...

当您想要拥有可选的路径参数时，这特别有用：

定义具有可选路径参数的路径的路由处理器¶

from litestar import get


@get(
   ["/some-path", "/some-path/{some_id:int}"],
)
async def my_route_handler(some_id: int = 1) -> None: ...

"保留" 关键字参数¶

路由处理器函数或方法通过将这些数据声明为带注解的函数关键字参数来访问各种数据。 Litestar 会检查带注解的关键字参数，然后将它们注入到请求处理器中。

可以使用带注解的函数关键字参数访问以下来源：

路径、查询、头和 cookie 参数
请求
注入的依赖项

此外，您可以指定以下特殊的 :term:`关键字参数 <argument>`（称为"保留关键字"）：

cookies: 将请求 cookies 作为解析后的 字典 注入。
headers: 将请求头作为解析后的 字典 注入。
query: 将请求 query_params 作为解析后的 字典 注入。
request: 注入 Request 实例。仅适用于 HTTP 路由处理器
scope: 注入 ASGI 作用域 字典。
socket: 注入 WebSocket 实例。仅适用于 websocket 路由处理器
state: 注入应用程序 State 的副本。
body: 原始请求体。仅适用于 HTTP 路由处理器

请注意，如果您的参数与上面任何保留的关键字参数冲突，您可以提供一个替代名称。

例如：

为保留关键字参数提供替代名称¶

from typing import Any, Dict
from litestar import Request, get
from litestar.datastructures import Headers, State


@get(path="/")
async def my_request_handler(
   state: State,
   request: Request,
   headers: Dict[str, str],
   query: Dict[str, Any],
   cookies: Dict[str, Any],
) -> None: ...

小技巧

您可以为应用程序状态定义自定义类型，然后将其用作类型，而不是仅使用 Litestar 的 State 类

类型注解¶

Litestar 强制执行严格的类型注解。由路由处理器装饰的函数必须对其所有参数和返回值进行类型注解。

如果缺少类型注解，将在应用程序启动过程中引发 ImproperlyConfiguredException。

强制执行此限制有几个原因：

确保最佳实践
确保一致的 OpenAPI 模式生成
允许 Litestar 在应用程序引导期间计算函数所需的参数

HTTP 路由处理器¶

HTTPRouteHandler 用于处理 HTTP 请求，可以使用 route() 装饰器创建：

使用 @route() 装饰器装饰函数来定义路由处理器¶

from litestar import HttpMethod, route


@route(path="/some-path", http_method=[HttpMethod.GET, HttpMethod.POST])
async def my_endpoint() -> None: ...

也可以不使用装饰器，直接使用 HTTPRouteHandler 来实现相同的效果：

通过创建 HTTPRouteHandler 实例来定义路由处理器¶

from litestar import HttpMethod
from litestar.handlers.http_handlers import HTTPRouteHandler


async def my_endpoint() -> None: ...

handler = HTTPRouteHandler(
    path="/some-path",
    http_method=[HttpMethod.GET, HttpMethod.POST],
    fn=my_endpoint
)

语义处理器装饰器 ¶

Litestar 还包括"语义" 装饰器，即预先设置了 http_method 关键字参数为特定 HTTP 动词的装饰器，这与它们的名称相关：

@delete()
@get()
@head()
@patch()
@post()
@put()

这些的使用方式与 @route() 完全相同，唯一的例外是您不需要配置 http_method 关键字参数：

此外，在 OpenAPI 规范中，HTTP 动词（例如 GET、POST 等）和路径的每个唯一组合都被视为一个独立的操作，每个操作都应通过唯一的 operation_id 来区分，并且最好还应有 summary 和 description 部分。

因此，不鼓励使用 @route() 装饰器。相反，首选模式是使用辅助类方法共享代码或将代码抽象为可重用的函数。

Websocket 路由处理器¶

可以使用 @websocket() 路由处理器来处理 WebSocket 连接。

备注

websocket 处理器是一种低级方法，需要直接处理套接字，并处理保持它打开、异常、客户端断开连接和内容协商。

有关处理 WebSocket 的更高级方法，请参阅 WebSockets

使用 @websocket() 路由处理器装饰器¶

from litestar import WebSocket, websocket


@websocket(path="/socket")
async def my_websocket_handler(socket: WebSocket) -> None:
   await socket.accept()
   await socket.send_json({...})
   await socket.close()

@websocket() 装饰器可用于创建 WebsocketRouteHandler 的实例。因此，以下代码与上面的代码等效：

直接使用 WebsocketRouteHandler 类¶

from litestar import WebSocket
from litestar.handlers.websocket_handlers import WebsocketRouteHandler

async def my_websocket_handler(socket: WebSocket) -> None:
   await socket.accept()
   await socket.send_json({...})
   await socket.close()

my_websocket_handler = WebsocketRouteHandler(
    path="/socket",
    fn=my_websocket_handler,
)

与 HTTP 路由处理器不同，websocket 处理器有以下要求：

它们必须声明一个 socket 关键字参数。
它们必须具有 None 的返回注解。
它们必须是 async 函数。

这些要求使用检查来强制执行，如果不满足任何一个要求，将引发有信息的异常。

OpenAPI 目前不支持 websockets。因此不会为这些路由处理器生成模式。

参见

ASGI 路由处理器¶

如果您需要编写自己的 ASGI 应用程序，可以使用 @asgi() 装饰器：

使用 @asgi() 路由处理器装饰器¶

from litestar.types import Scope, Receive, Send
from litestar.status_codes import HTTP_400_BAD_REQUEST
from litestar import Response, asgi


@asgi(path="/my-asgi-app")
async def my_asgi_app(scope: Scope, receive: Receive, send: Send) -> None:
   if scope["type"] == "http":
       if scope["method"] == "GET":
           response = Response({"hello": "world"})
           await response(scope=scope, receive=receive, send=send)
       return
   response = Response(
       {"detail": "unsupported request"}, status_code=HTTP_400_BAD_REQUEST
   )
   await response(scope=scope, receive=receive, send=send)

@asgi() 装饰器可用于创建 ASGIRouteHandler 的实例。因此，以下代码与上面的代码等效：

直接使用 ASGIRouteHandler 类¶

from litestar import Response
from litestar.handlers.asgi_handlers import ASGIRouteHandler
from litestar.status_codes import HTTP_400_BAD_REQUEST
from litestar.types import Scope, Receive, Send

async def my_asgi_app(scope: Scope, receive: Receive, send: Send) -> None:
   if scope["type"] == "http":
       if scope["method"] == "GET":
           response = Response({"hello": "world"})
           await response(scope=scope, receive=receive, send=send)
       return
   response = Response(
       {"detail": "unsupported request"}, status_code=HTTP_400_BAD_REQUEST
   )
   await response(scope=scope, receive=receive, send=send)

my_asgi_app = ASGIRouteHandler(path="/my-asgi-app", fn=my_asgi_app)

ASGI 路由处理器注意事项¶

与其他路由处理器不同，@asgi() 路由处理器只接受三个必须定义的关键字参数：

scope，描述 ASGI 连接的值映射。它总是包含一个 type 键，值为 http 或 websocket，以及一个 path 键。如果类型是 http，scope 字典还将包括一个 method 键，值为 DELETE、GET、POST、PATCH、PUT、HEAD 之一。
receive，ASGI 应用程序接收消息的注入函数。
send，ASGI 应用程序发送消息的注入函数。

您可以在 ASGI 规范中阅读更多相关信息。

此外，ASGI 路由处理器函数必须是 async 函数。这使用检查来强制执行，如果函数不是 async 函数，将引发有信息的异常。

有关 @asgi() 装饰器及其接受的关键字参数的完整详细信息，请参阅 ASGIRouteHandler API 参考文档。

路由处理器索引¶

您可以在所有路由处理器装饰器中提供一个 name 关键字参数。此关键字参数的值 必须是唯一的，否则将引发 ImproperlyConfiguredException 异常。

name 的默认值是处理器的 __str__() 方法返回的值，它应该是处理器的完整点分路径（例如，对于位于 app/controllers/projects.py 文件中的 list 函数，应该是 app.controllers.projects.list）。 name 可用于在运行时动态检索包含路由处理器实例和路径的映射，也可用于为该处理器构建 URL 路径：

使用 name 关键字参数检索路由处理器实例和路径¶

from litestar import Litestar, Request, get
from litestar.exceptions import NotFoundException
from litestar.response import Redirect


@get("/abc", name="one")
def handler_one() -> None:
    pass


@get("/xyz", name="two")
def handler_two() -> None:
    pass


@get("/def/{param:int}", name="three")
def handler_three(param: int) -> None:
    pass


@get("/{handler_name:str}", name="four")
def handler_four(request: Request, name: str) -> Redirect:
    handler_index = request.app.get_handler_index_by_name(name)
    if not handler_index:
        raise NotFoundException(f"no handler matching the name {name} was found")

    # handler_index == { "paths": ["/"], "handler": ..., "qualname": ... }
    # 在下面对处理器索引做一些操作，例如发送重定向响应到处理器，或访问
    # handler.opt 和存储在那里的一些值等。

    return Redirect(path=handler_index[0])


@get("/redirect/{param_value:int}", name="five")
def handler_five(request: Request, param_value: int) -> Redirect:
    path = request.app.route_reverse("three", param=param_value)
    return Redirect(path=path)


app = Litestar(route_handlers=[handler_one, handler_two, handler_three])

如果未找到具有给定名称的路由，或者缺少任何路径参数，或者传递的任何路径参数类型与相应路由声明中的类型不匹配， route_reverse() 将引发 NoRouteMatchFoundException。

但是，str 可以代替 datetime、date、 time、timedelta、float 和 Path 参数被接受，因此您可以应用自定义格式并将结果传递给 route_reverse()。

如果处理器附加了多个路径，route_reverse() 将返回消耗传递给函数的关键字参数数量最多的路径。

使用 route_reverse() 方法为路由处理器构建 URL 路径¶

from litestar import get, Request


@get(
   ["/some-path", "/some-path/{id:int}", "/some-path/{id:int}/{val:str}"],
   name="handler_name",
)
def handler(id: int = 1, val: str = "default") -> None: ...


@get("/path-info")
def path_info(request: Request) -> str:
   path_optional = request.app.route_reverse("handler_name")
   # /some-path`

   path_partial = request.app.route_reverse("handler_name", id=100)
   # /some-path/100

   path_full = request.app.route_reverse("handler_name", id=100, val="value")
   # /some-path/100/value`

   return f"{path_optional} {path_partial} {path_full}"

当处理器与具有相同路径参数的多个路由相关联时（例如，在多个路由器上注册的索引处理器），route_reverse() 的输出是不可预测的。此 callable 将返回格式化的路径；但是，其选择可能看起来是任意的。因此，强烈建议不要在这些条件下反转 URL。

如果您有权访问 Request 实例，您可以使用 url_for() 方法进行反向查找，该方法类似于 route_reverse()，但返回绝对 URL。

向处理器添加任意元数据¶

所有路由处理器装饰器都接受一个名为 opt 的键，它接受一个字典，其中包含任意值，例如：

通过 opt 关键字参数向路由处理器添加任意元数据¶

from litestar import get


@get("/", opt={"my_key": "some-value"})
def handler() -> None: ...

此字典可以由路由守卫访问，或通过访问 Request 对象上的 route_handler 属性，或直接使用 ASGI scope 对象访问。

基于 opt，您可以将任何任意关键字参数传递给路由处理器装饰器，它将自动设置为 opt 字典中的键：

通过 opt 关键字参数向路由处理器添加任意元数据¶

from litestar import get

@get("/", my_key="some-value")
def handler() -> None: ...

assert handler.opt["my_key"] == "some-value"

您可以在应用程序的所有层指定 opt 字典。在特定路由处理器、控制器、路由器上，甚至在应用程序实例本身上，如分层架构中所述。

生成的字典是通过合并所有层的 opt 字典构造的。如果多个层定义相同的键，则离响应处理器最近的层的值将优先。

签名命名空间 ¶

Litestar 为任何处理器或依赖函数的参数生成一个模型，称为"签名模型"，用于解析和验证要注入函数的原始数据。

构建模型需要在运行时检查签名参数的名称和类型，因此类型必须在模块的作用域内可用 - 这是 linting 工具（如 ruff 或 flake8-type-checking）会主动监视并提出建议的内容。

例如，在以下代码段中，名称 Model 在运行时不可用：

具有在运行时不可用的类型的路由处理器¶

from __future__ import annotations

from typing import TYPE_CHECKING

from litestar import Controller, post

if TYPE_CHECKING:
    from domain import Model


class MyController(Controller):
    @post()
    def create_item(data: Model) -> Model:
        return data

在此示例中，Litestar 将无法生成签名模型，因为类型 Model 在运行时不存在于模块作用域中。我们可以通过逐个情况消除 linter，例如：

然而，这种方法可能会变得乏味；作为替代方案，Litestar 在应用程序的每个层上接受一个 signature_types 序列，如以下示例所示：

此模块定义了我们的控制器，请注意，我们不将 Model 导入到运行时命名空间，也不需要任何指令来控制 linter 的行为。

最后，我们确保我们的应用程序知道，当它在解析签名时遇到名称"Model"时，它应该引用我们的域 Model 类型。

小技巧

如果您想将类型映射到与其 __name__ 属性不同的名称，您可以使用 signature_namespace 参数，例如，app = Litestar(signature_namespace={"FooModel": Model})。

这使得可以在 if TYPE_CHECKING 块内使用类似 from domain.foo import Model as FooModel 的导入模式。

默认签名命名空间 ¶

Litestar 在解析签名模型时自动将一些名称添加到签名命名空间，以支持 "保留" 关键字参数的注入。

这些名称是：

Headers
ImmutableState
Receive
Request
Scope
Send
State
WebSocket
WebSocketScope

这些名称中的任何一个的导入都可以安全地留在 if TYPE_CHECKING: 块中，无需任何配置。

路由处理器¶

声明路径¶

"保留" 关键字参数¶

类型注解¶

HTTP 路由处理器¶

语义处理器 装饰器¶

Websocket 路由处理器¶

ASGI 路由处理器¶

ASGI 路由处理器注意事项¶

路由处理器索引¶

向处理器添加任意元数据¶

签名 命名空间¶

默认签名 命名空间¶

语义处理器装饰器 ¶

签名命名空间 ¶

默认签名命名空间 ¶