Skip to content

请求对象 ​

请求对象是一个数据类,包含了与该次请求相关的所有信息。它会在路由处理程序中作为第一个参数传入。

该请求对象在 Rust 端创建,但会作为数据类暴露给 Python。

  • Attributes:
  • query_params (QueryParams):请求的查询参数。`例如:/user?id=123 -> {"id": ["123"]}`
  • headers (dict[str, str]):请求的标头。`例如:{"Content-Type": "application/json"}`
  • params (dict[str, str]):请求的路径参数。`例如:/user/:id -> {"id": "123"}`
  • body (Union[str, bytes]):请求的原始正文。对于 JSON 请求体,请使用 `json()` 方法将正文解析为具有正确类型保留的字典。
  • method (str):请求的方法。`例如:GET、POST、PUT、DELETE`
  • ip_addr (Optional[str]):客户端的 IP 地址
  • identity (Optional[Identity]):客户端的身份
py
@dataclass
class Request:
  """
  query_params
  headers
  path_params
  body
  method
  url
  form_data
  files
  ip_addr
  identity
  """
py
@dataclass
class Request:
  """
  query_params: QueryParams
  headers: Headers
  path_params: dict[str, str]
  body: Union[str, bytes]
  method: str
  url: Url
  form_data: dict[str, str]
  files: dict[str, bytes]
  ip_addr: Optional[str]
  identity: Optional[Identity]
  """

解析 JSON 请求体 ​

request.json() 方法将请求体解析为 JSON,并返回一个保留完整类型的 Python dict 或 list(JSON 对象返回 dict,JSON 数组返回 list):

  • JSON null 转换为 Python None
  • JSON 数字转换为 Python int 或 float
  • JSON 布尔值转换为 Python bool
  • JSON 字符串转换为 Python str
  • JSON 数组转换为 Python list
  • JSON 对象转换为 Python dict

嵌套结构将被递归处理,最大深度为 128 层。

py
@app.post("/example")
async def handler(request):
    data = request.json()  # 返回一个保留类型的字典或列表
    # 例如 {"count": 42, "active": true, "tags": ["a", "b"]}
    # ->   {"count": 42, "active": True, "tags": ["a", "b"]}
    return {"received": data}
py
@app.post("/example")
async def handler(request: Request):
    data = request.json()  # 返回一个保留类型的字典或列表
    # 例如 {"count": 42, "active": true, "tags": ["a", "b"]}
    # ->   {"count": 42, "active": True, "tags": ["a", "b"]}
    return {"received": data}

如果请求体不是有效的 JSON,将会抛出 ValueError。

Extra Path Parameters ​

Robyn 支持通过 *extra 语法捕获额外的路径参数,这样可以捕获在定义的路由之后的所有额外的路径段。

例如,如果有这样一个路由:

py
@app.get("/sync/extra/*extra")
def sync_param_extra(request):
    extra = request.path_params["extra"]
    return extra
py
@app.get("/sync/extra/*extra")
def sync_param_extra(request: Request):
    extra = request.path_params["extra"]
    return extra

在 /sync/extra/ 后的任何路径段都会被捕获到 extra 参数中。例如:

  • 请求 `/sync/extra/foo/bar` 将使得 `extra = "foo/bar`"
  • 请求 `/sync/extra/123/456/789` 将使得 `extra = "123/456/789`"

你可以通过 request.path_params["extra"] 在路由处理程序中访问这些额外的路径参数。

在处理动态嵌套路由,或者捕获未知数量的路径段时,这个功能将有奇效。


便捷参数访问 ​

你可以在函数签名中直接声明带类型注解的查询参数和路径参数,无需手动从请求对象中提取和转换。Robyn 会自动解析并进行类型转换。

任何不匹配已知请求组件(Request、QueryParams、Headers 等)的处理函数参数,都会被视为独立的路径参数或查询参数。

基本用法 — 带类型转换和默认值的路径参数与查询参数。

py
@app.get("/items/:id")
async def get_item(id, q, page= 1):
    # id 从路径参数字符串自动转换为 int
    # q 从 ?q=... 获取
    # page 未提供时默认为 1
    return {"id": id, "q": q, "page": page}
py
@app.get("/items/:id")
async def get_item(id: int, q: str, page: int = 1):
    # id 从路径参数字符串自动转换为 int
    # q 从 ?q=... 获取
    # page 未提供时默认为 1
    return {"id": id, "q": q, "page": page}
py
@app.get("/items/:id")
async def get_item(request, id, q= ""):
    # request 仍然按原来的方式注入
    # id 和 q 作为独立参数解析
    return {"id": id, "q": q, "method": request.method}
py
@app.get("/items/:id")
async def get_item(request: Request, id: int, q: str = ""):
    # request 仍然按原来的方式注入
    # id 和 q 作为独立参数解析
    return {"id": id, "q": q, "method": request.method}

Optional、List、Bool 和 Float 参数 — Robyn 自动处理常见的 Python 类型。

  • Optional[T] — 未提供时解析为 None
  • List[T] — 收集重复的查询参数(例如 ?tag=a&tag=b)
  • bool — 接受 true/false、1/0、yes/no、on/off
  • float — 标准浮点数转换
py
@app.get("/search")
def search(name, age= None):
    return {"name": name, "age": age}
## GET /search?name=bob        -> {"name": "bob", "age": null}
# GET /search?name=bob&age=30 -> {"name": "bob", "age": 30}
py
@app.get("/search")
def search(name: str, age: Optional[int] = None):
    return {"name": name, "age": age}
# GET /search?name=bob        -> {"name": "bob", "age": null}
# GET /search?name=bob&age=30 -> {"name": "bob", "age": 30}
py
@app.get("/filter")
def filter_items(tag):
    return {"tags": tag}
# GET /filter?tag=python&tag=rust -> {"tags": ["python", "rust"]}
py
from typing import List

@app.get("/filter")
def filter_items(tag: List[str]):
    return {"tags": tag}
# GET /filter?tag=python&tag=rust -> {"tags": ["python", "rust"]}
py
@app.get("/settings")
def settings(active= False, price= 0.0):
    return {"active": active, "price": price}
# GET /settings?active=true&price=19.99
# -> {"active": true, "price": 19.99}
py
@app.get("/settings")
def settings(active: bool = False, price: float = 0.0):
    return {"active": active, "price": price}
# GET /settings?active=true&price=19.99
# -> {"active": true, "price": 19.99}

错误处理 — 如果缺少必需的参数或值无法转换为声明的类型,Robyn 会自动返回 400 Bad Request 响应。

py
@app.get("/items/:id")
def get_item(id, q):
    return {"id": id, "q": q}

# GET /items/42         -> 400(缺少必需的 'q')
# GET /items/abc?q=test -> 400(无法将 'abc' 转换为 int)
py
@app.get("/items/:id")
def get_item(id: int, q: str):
    return {"id": id, "q": q}

# GET /items/42         -> 400(缺少必需的 'q')
# GET /items/abc?q=test -> 400(无法将 'abc' 转换为 int)

类型别名 ​

Robyn 提供了一组便捷的类型别名,用于描述请求各组成部分的运行时类型,这样你就可以为处理函数添加类型注解,并通过 mypy/pyright 进行类型检查。它们既可以从 robyn 导入,也可以从 robyn.types 导入。

  • `RequestMethod`:`Literal["GET", "POST", "PUT", "DELETE", "PATCH", "HEAD", "OPTIONS", "CONNECT", "TRACE"]` —— 即 `request.method` 的取值。
  • `RequestBody`:`str | bytes` —— 即 `request.body` 的类型(UTF-8 文本为 `str`,否则为 `bytes`)。
  • `RequestURL`:`Url` 的别名 —— 即 `request.url` 的类型。
py
from robyn import Robyn

app = Robyn(__file__)

@app.get("/typed")
def handler(request):
    method= request.method
    body= request.body
    url= request.url
    return {"method": method}
py
from robyn import Robyn, Request
from robyn import RequestMethod, RequestBody, RequestURL

app = Robyn(__file__)

@app.get("/typed")
def handler(request: Request):
    method: RequestMethod = request.method
    body: RequestBody = request.body
    url: RequestURL = request.url
    return {"method": method}

下一步 ​

接下来,蝙蝠侠希望了解 Robyn 服务器的配置。于是他开始了解 Robyn 环境配置文件的概念。

基于 MIT 许可发布