快速开始 | Robyn 罗宾

同步与异步请求

首先，Robyn 向蝙蝠侠介绍了它处理同步和异步请求的能力。蝙蝠侠很高兴了解这些功能，并开始在应用中实现它们。

对于一个简单的同步请求，蝙蝠侠使用以下代码：

untypedtyped

from robyn import Robyn

app = Robyn(__file__)

@app.get("/")
def h(request):
    return "Hello, world"

app.start(port=8080, host="0.0.0.0") # host 是可选的，默认为 127.0.0.1

from robyn import Robyn, Request

app = Robyn(__file__)

@app.get("/")
def h(request: Request):
    return "Hello, world"

app.start(port=8080, host="0.0.0.0") # host 是可选的，默认为 127.0.0.1

对于异步请求，蝙蝠侠使用了以下代码：

untypedtyped

@app.get("/")
async def h(request):
    return "Hello, world"

from robyn import Request

@app.get("/")
async def h(request: Request) -> str:
    return "Hello, world"

运行 Robyn

蝙蝠侠对如何运行该应用程序感到好奇。Robyn 解释说，他可以通过一个简单的命令 python3 app.py 来运行应用程序 app.py。

Robyn 公开如下命令。这些命令可用于运行应用程序或生成新项目。

bash

usage: app.py [-h] [--processes PROCESSES] [--workers WORKERS] [--log-level LOG_LEVEL] [--create] [--docs] [--open-browser] [--version]

Robyn，一个快速的异步 web 框架，基于 Rust 运行。

选项:
  -h, --help            显示帮助信息并退出
  --processes PROCESSES
                        选择进程的数量。[默认值: 1]
  --workers WORKERS     选择工作线程的数量。[默认值: 1]
  --dev                 开发者模式。它会根据文件更改重新启动服务器。
  --log-level LOG_LEVEL
                        设置日志级别
  --create              创建一个新的项目模板。
  --docs                打开 Robyn 文档。
  --open-browser        启动成功后打开浏览器。
  --version             显示 Robyn 的版本。
  --compile-rust-path COMPILE_RUST_PATH
                        编译指定路径下的 Rust 文件。
  --create-rust-file CREATE_RUST_FILE
                        创建一个指定名称的 Rust 文件。
  --disable-openapi     禁用 OpenAPI 文档。
  --fast                快速模式。它设置进程、工作线程和日志级别的最佳值。不过，您可以覆盖这些设置。

另外，您还可以使用 Robyn 的 CLI 来运行应用程序，即 python -m robyn app.py。

bash

usage: python -m robyn app.py [-h] [--processes PROCESSES] [--workers WORKERS] [--dev] [--log-level LOG_LEVEL] [--create] [--docs] [--open-browser] [--version]

Robyn，一个快速的异步 web 框架，基于 Rust 运行。

选项:
  -h, --help            显示帮助信息并退出
  --processes PROCESSES
                        选择进程的数量。[默认值: 1]
  --workers WORKERS     选择工作线程的数量。[默认值: 1]
  --dev                 开发者模式。它会根据文件更改重新启动服务器。
  --log-level LOG_LEVEL
                        设置日志级别
  --create              创建一个新的项目模板。
  --docs                打开 Robyn 文档。
  --open-browser        启动成功后打开浏览器。
  --version             显示 Robyn 的版本。
  --compile-rust-path COMPILE_RUST_PATH
                        编译指定路径下的 Rust 文件。
  --create-rust-file CREATE_RUST_FILE
                        创建一个指定名称的 Rust 文件。
  --disable-openapi     禁用 OpenAPI 文档。
  --fast                快速模式。它设置进程、工作线程和日志级别的最佳值。不过，您可以覆盖这些设置。

处理 HTTP 请求

然后，Robyn 教蝙蝠侠如何处理各种 HTTP 请求，如 GET、POST、PUT、PATCH 和 DELETE。在 Robyn 的指导下，蝙蝠侠可以为每种请求类型创建接口，从而使应用程序更加灵活和高效。

例如，蝙蝠侠学会了创建如下 POST 请求：

untypedtyped

@app.post("/")
async def h(request):
    return "Hello World"

from robyn import Request

@app.post("/")
async def h(request: Request):
    return "Hello World"

响应 JSON 数据

蝙蝠侠对从应用程序返回 JSON 响应的能力感到好奇。Robyn 向他展示了如何使用 jsonify 函数来实现这一点。

现在，蝙蝠侠可以从应用程序返回 JSON 响应，从而方便前端解析数据。

untypedtyped

from robyn import jsonify

@app.post("/jsonify")
async def json(request):
    return {"hello": "world"}

from robyn import jsonify, Request

@app.post("/jsonify")
async def json(request: Request):
    return {"hello": "world"}

获取路径参数与查询参数

蝙蝠侠想了解如何从传入的请求中访问路径参数和查询参数，这样他就能创建动态路由，并从请求中提取特定信息。

Robyn 向蝙蝠侠展示了如何从请求中访问路径参数和查询参数。以下是相关的示例：

例如，蝙蝠侠可以创建带路径参数的路由，并通过以下代码访问该参数：

untypedtyped

from robyn import jsonify

@app.post("/jsonify/:id")
async def json(req_obj, path_parameters):
    print(req_obj.path_params["id"])
    print(path_parameters["id"])
    assert req_obj.path_params["id"] == path_parameters["id"]
    return {"hello": "world"}

from robyn import jsonify
from robyn.types import PathParams

@app.post("/jsonify/:id")
async def json(req_obj: Request, path_parameters: PathParams):
    print(req_obj.path_params["id"])
    print(path_parameters["id"])
    assert req_obj.path_params["id"] == path_parameters["id"]
    return {"hello": "world"}

要访问查询参数，蝙蝠侠可以使用以下代码：

untypedtyped

@app.get("/query")
async def query_get(req_obj, query_params):
    query_data = query_params.to_dict()
    assert query_data == req_obj.query_params.to_dict()
    return jsonify(query_data)

from robyn import Request
from robyn.robyn import QueryParams

@app.get("/query")
async def query_get(req_obj: Request, query_params: QueryParams):
    query_data = query_params.to_dict()
    assert query_data == req_obj.query_params.to_dict()
    return jsonify(query_data)

任何请求参数都可以在处理程序函数中通过类型注释或使用保留名称进行访问。

请注意，类型注释会优先于保留名称。 Robyn 向蝙蝠侠演示了访问请求参数的不同语法示例：

untypedtyped

from robyn.types import PathParams

@app.get("/untyped/query_params")
def untyped_basic(query_params):
    return query_params.to_dict()

@app.get("/typed/query_params")
def typed_basic(query_data):
    return query_data.to_dict()

@app.get("/untyped/path_params/:id")
def untyped_path_params(query_params):
    return query_params  # 由于类型注释优先，PathParams 包含路径参数

@app.post("/typed_untyped/combined")
def typed_untyped_combined(
        query_params,
        method_data,
        body_data,
        url,
        headers_item,
):
    return {
        "body": body_data,
        "query_params": query_params.to_dict(),
        "method": method_data,
        "url": url.path,
        "headers": headers_item.get("server"),
    }

from robyn.robyn import QueryParams, Headers
from robyn.types import PathParams, RequestMethod, RequestBody, RequestURL

@app.get("/untyped/query_params")
def untyped_basic(query_params):
    return query_params.to_dict()

@app.get("/typed/query_params")
def typed_basic(query_data: QueryParams):
    return query_data.to_dict()

@app.get("/untyped/path_params/:id")
def untyped_path_params(query_params: PathParams):
    return query_params  # 由于类型注释优先，PathParams 包含路径参数

@app.post("/typed_untyped/combined")
def typed_untyped_combined(
        query_params,
        method_data: RequestMethod,
        body_data: RequestBody,
        url: RequestURL,
        headers_item: Headers,
):
    return {
        "body": body_data,
        "query_params": query_params.to_dict(),
        "method": method_data,
        "url": url.path,
        "headers": headers_item.get("server"),
    }

类型别名: Request、QueryParams、Headers、PathParams、RequestBody、RequestURL、FormData、RequestFiles、RequestIP、RequestIdentity

保留名称: r、req、request、query_params、headers、path_params、body、method、url、ip_addr、identity、form_data、files

随着蝙蝠侠继续使用 Robyn 开发 Web 应用，他探索了更多功能，并通过代码示例实现它们。

自定义响应格式与响应头

在了解了 Robyn 的动态特性后，蝙蝠侠想要自定义响应格式和响应头。Robyn 向他展示了如何使用字典和 Response 对象来实现这一功能。

使用字典

蝙蝠侠学会了通过返回字典或使用 Robyn 的 Response 对象来定制响应格式，并为每个响应设置状态码和响应头。下面是一个使用字典创建响应的示例：

untypedtyped

@app.post("/dictionary")
async def dictionary(request):
    return {
        "status_code": 200,
        "description": "This is a regular response",
        "type": "text",
        "headers": {"Header": "header_value"},
    }

from robyn import Request

@app.post("/dictionary")
async def dictionary(request: Request):
    return {
        "status_code": 200,
        "description": "This is a regular response",
        "type": "text",
        "headers": {"Header": "header_value"},
    }

使用 Response 对象

蝙蝠侠还学会了如何使用 Response 对象，示例如下：

untypedtyped

from robyn.robyn import Response

@app.get("/response")
async def response(request):
    return Response(status_code=200, headers=Headers({}), description="OK")

from robyn.robyn import Response, Request

@app.get("/response")
async def response(request: Request):
    return Response(status_code=200, headers=Headers({}), description="OK")

返回二进制输出

蝙蝠侠还希望从应用程序返回二进制数据。他可以通过将响应类型设置为 binary 并返回一个字节对象来实现：

untypedtyped

from robyn import Response

@app.get("/binary_output_response_sync")
def binary_output_response_sync(request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

@app.get("/binary_output_async")
async def binary_output_async(request):
    return b"OK"

@app.get("/binary_output_response_async")
async def binary_output_response_async(request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

from robyn import Request, Response

@app.get("/binary_output_response_sync")
def binary_output_response_sync(request: Request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

@app.get("/binary_output_async")
async def binary_output_async(request: Request):
    return b"OK"

@app.get("/binary_output_response_async")
async def binary_output_response_async(request: Request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

响应头

作为世界上最伟大的侦探，蝙蝠侠在 Response 对象中发现了 headers 字段，他希望进一步了解如何使用它来设置响应头。例如，蝙蝠侠可以通过以下方式设置 Content-Type 响应头为 application/json：

局部响应头

通过使用 Response 对象中的 headers 字段：

untypedtyped

@app.get("/")
def binary_output_response_sync(request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

from robyn import Request

@app.get("/")
def binary_output_response_sync(request: Request):
    return Response(
        status_code=200,
        headers={"Content-Type": "application/octet-stream"},
        description="OK",
    )

全局响应头

蝙蝠侠还可以为所有路由器设置全局响应头：

untypedtyped

app.add_response_header("content-type", "application/json")

app.add_response_header("content-type", "application/json")

add_response_header 会将响应头附加到响应头列表中，而 set_response_header 会替换现有响应头（如果有的话）。

untypedtyped

app.set_response_header("content-type", "application/json")

app.set_response_header("content-type", "application/json")

如果希望某些接口不使用响应头，可以使用 exclude_response_headers_for 函数。

untypedtyped

app.exclude_response_headers_for(["/login", "/signup"])

app.exclude_response_headers_for(["/login", "/signup"])

Cookies

Robyn 提供了符合 RFC 6265 标准的完整 Cookie API。使用 Response 对象上的 set_cookie 方法来设置 Cookie。

untypedtyped

from robyn import Response, Headers

@app.get("/")
def set_session(request):
    response = Response(200, Headers({}), "Welcome!")
    response.set_cookie(key="session", value="abc123")
    return response

from robyn import Request, Response, Headers

@app.get("/")
def set_session(request: Request):
    response = Response(200, Headers({}), "Welcome!")
    response.set_cookie(key="session", value="abc123")
    return response

您可以设置额外的 Cookie 属性以增强安全性和控制：

path: Cookie 路径（默认："/"）
domain: Cookie 域名
max_age: Cookie 有效期（秒）
secure: 仅通过 HTTPS 发送
http_only: JavaScript 无法访问
same_site: CSRF 保护（"Strict"、"Lax" 或 "None" - 不区分大小写）

untypedtyped

from robyn import Response, Headers

@app.get("/login")
def login(request):
    response = Response(200, Headers({}), "Logged in")
    response.set_cookie(
        key="auth_token",
        value="secret123",
        path="/",
        max_age=3600,        # 1 小时
        secure=True,         # 仅 HTTPS
        http_only=True,      # 禁止 JavaScript 访问
        same_site="Strict",  # CSRF 保护
    )
    return response

from robyn import Request, Response, Headers

@app.get("/login")
def login(request: Request):
    response = Response(200, Headers({}), "Logged in")
    response.set_cookie(
        key="auth_token",
        value="secret123",
        path="/",
        max_age=3600,        # 1 小时
        secure=True,         # 仅 HTTPS
        http_only=True,      # 禁止 JavaScript 访问
        same_site="Strict",  # CSRF 保护
    )
    return response

要从浏览器删除 Cookie，请使用 cookies 集合上的 delete 方法。这会设置 max_age=0，告诉浏览器删除该 Cookie。

untypedtyped

from robyn import Response, Headers

@app.get("/logout")
def logout(request):
    response = Response(200, Headers({}), "Logged out")
    response.cookies.delete("auth_token")
    return response

from robyn import Request, Response, Headers

@app.get("/logout")
def logout(request: Request):
    response = Response(200, Headers({}), "Logged out")
    response.cookies.delete("auth_token")
    return response

您可以遍历 Cookie 或按名称访问它们：

untypedtyped

from robyn import Response, Headers

@app.get("/debug")
def debug_cookies(request):
    response = Response(200, Headers({}), "Cookies set")
    response.set_cookie("a", "1")
    response.set_cookie("b", "2")

    # 获取所有 Cookie 名称
    names = response.cookies.keys()

    # 遍历 Cookie
    for name in response.cookies:
        print(f"Cookie: {name}")

    # 检查 Cookie 是否存在
    if "a" in response.cookies:
        print("Cookie 'a' exists")

    return response

from robyn import Request, Response, Headers

@app.get("/debug")
def debug_cookies(request: Request):
    response = Response(200, Headers({}), "Cookies set")
    response.set_cookie("a", "1")
    response.set_cookie("b", "2")

    # 获取所有 Cookie 名称
    names = response.cookies.keys()

    # 遍历 Cookie
    for name in response.cookies:
        print(f"Cookie: {name}")

    # 检查 Cookie 是否存在
    if "a" in response.cookies:
        print("Cookie 'a' exists")

    return response

请求头

蝙蝠侠现在想了解如何读取请求头。Robyn 向他解释，他可以使用 request.headers 字段来读取响应头。例如，蝙蝠侠可以通过以下方式读取 Content-Type 请求头：

局部请求头

通过使用 Request 对象中的 headers 字段：

untypedtyped

@app.get("/")
def binary_output_response_sync(request):
  headers = request.headers

  print("These are the request headers: ", headers)
  existing_header = headers.get("exisiting_header")
  existing_header = headers.get("exisiting_header", "default_value")
  exisiting_header = headers["exisiting_header"] # This syntax is also valid

  headers.set("modified", "modified_value")
  headers["new_header"] = "new_value" # This syntax is also valid

  print("These are the modified request headers: ", headers)

  return ""

from robyn import Request

@app.get("/")
def binary_output_response_sync(request: Request):
  headers = request.headers

  print("These are the request headers: ", headers)
  existing_header = headers.get("exisiting_header")
  existing_header = headers.get("exisiting_header", "default_value")
  exisiting_header = headers["exisiting_header"] # This syntax is also valid

  headers.set("modified", "modified_value")
  headers["new_header"] = "new_value" # This syntax is also valid

  print("These are the modified request headers: ", headers)

  return ""

或者使用全局请求头：

untypedtyped

app.add_request_header("server", "robyn")

app.add_request_header("server", "robyn")

add_request_header 会将请求头附加到请求头列表中，而 set_request_header 会替换现有请求头（如果有的话）。

untypedtyped

app.set_request_header("server", "robyn")

app.set_request_header("server", "robyn")

响应状态码

了解了响应格式和响应头之后，蝙蝠侠学会了为他的响应设置状态码：

untypedtyped

from robyn import status_codes

@app.get("/response")
async def response(request):
    return Response(status_code=status_codes.HTTP_200_OK, headers=Headers({}), description="OK")

from robyn import status_codes, Request

@app.get("/response")
async def response(request: Request):
    return Response(status_code=status_codes.HTTP_200_OK, headers=Headers({}), description="OK")

下一步

接下来，蝙蝠侠想知道 Robyn 提到的 Request 请求对象是什么。Robyn 说：“下一部分。”

请求对象

蝙蝠侠也对 Robyn 的架构感兴趣。Robyn 继续说道：“下一部分。”

架构

同步与异步请求 ​

运行 Robyn ​

处理 HTTP 请求 ​

响应 JSON 数据 ​

获取路径参数与查询参数 ​

自定义响应格式与响应头 ​

使用字典 ​

使用 Response 对象 ​

返回二进制输出 ​

响应头 ​

局部响应头 ​

全局响应头 ​

Cookies ​

Cookie 属性 ​

删除 Cookie ​

访问 Cookie ​

请求头 ​

局部请求头 ​

响应状态码 ​

下一步 ​

同步与异步请求

运行 Robyn

处理 HTTP 请求

响应 JSON 数据

获取路径参数与查询参数

自定义响应格式与响应头

使用字典

使用 Response 对象

返回二进制输出

响应头

局部响应头

全局响应头

Cookies

Cookie 属性

删除 Cookie

访问 Cookie

请求头

局部请求头

响应状态码

下一步