Robyn 罗宾

响应返回样式 ​

Robyn 会自动将处理器的返回值转换为合适的 HTTP 响应。下面是支持的所有返回样式的完整参考。

字符串 ​

最简单的返回样式。以状态码 200 返回字符串，并将其作为 text/plain，字符串会自动以 UTF-8 编码。

py
from robyn import Robyn

app = Robyn(__file__)

@app.get("/hello")
def hello():
    return "Hello, World!"
py
from robyn import Robyn

app = Robyn(__file__)

@app.get("/hello")
def hello():
    return "Hello, World!"

字典或列表 ​

返回 dict 或 list 会自动序列化为 JSON，并设置 Content-Type: application/json，状态码为 200。

py
@app.get("/user")
def get_user():
    return {"name": "Alice", "age": 30}
py
@app.get("/user")
def get_user():
    return {"name": "Alice", "age": 30}
py
@app.get("/items")
def get_items():
    return [1, 2, 3, "four", {"nested": True}]
py
@app.get("/items")
def get_items():
    return [1, 2, 3, "four", {"nested": True}]

Response 对象 ​

Response 对象让你完全控制状态码、响应头和主体。在需要对响应进行全面自定义时使用它。

响应主体既可以通过 body（推荐使用、语义更直观的参数）传入，也可以通过 description（为向后兼容保留的旧名称）传入。两者互为别名 —— 只传其中一个。headers 是可选的，默认为空的 Headers 对象。

  • `status_code` (int)：HTTP 状态码，例如 `200`。
  • `headers` (Headers | dict | None)：响应头。接受 `Headers` 实例、普通 `dict` 或 `None`（表示为空）。
  • `body` (str | bytes)：响应主体。`description` 的别名。
  • `description` (str | bytes)：响应主体（旧名称，建议改用 `body`）。
py
from robyn import Robyn, Response, Headers

app = Robyn(__file__)

@app.get("/custom")
async def custom_response():
    return Response(
        status_code=200,
        headers=Headers({"X-Custom": "value"}),
        body="OK",
    )
py
from robyn import Robyn, Response, Headers

app = Robyn(__file__)

@app.get("/custom")
async def custom_response():
    return Response(
        status_code=200,
        headers=Headers({"X-Custom": "value"}),
        body="OK",
    )
py
@app.get("/custom")
async def custom_response():
    # `description` 仍然有效，与 `body` 完全等价
    return Response(status_code=200, description="OK")
py
@app.get("/custom")
async def custom_response():
    # `description` 仍然有效，与 `body` 完全等价
    return Response(status_code=200, description="OK")

Bytes ​

返回 bytes 对象会设置 Content-Type: application/octet-stream，状态码为 200。适用于图像或在内存中生成的文件等二进制数据。

py
@app.get("/binary")
async def binary_data():
    return b"binary data"
py
@app.get("/binary")
async def binary_data():
    return b"binary data"

Pydantic BaseModel ​

如果返回一个 Pydantic 的 BaseModel 实例，Robyn 会自动将其序列化为 JSON，并设置 Content-Type: application/json 和状态码 200。

注意： Pydantic 必须作为可选依赖安装（pip install pydantic）。如果未安装，模型将无法被检测，处理将回退到默认的字符串序列化。

py
from pydantic import BaseModel
from robyn import Robyn

app = Robyn(__file__)

class User(BaseModel):
    name
    email
    age

@app.get("/model")
def get_model():
    return User(name="Alice", email="[email protected]", age=30)
py
from pydantic import BaseModel
from robyn import Robyn

app = Robyn(__file__)

class User(BaseModel):
    name: str
    email: str
    age: int

@app.get("/model")
def get_model():
    return User(name="Alice", email="[email protected]", age=30)

元组 (body, headers, status_code) ​

一个三元素元组 (body, headers, status_code) 允许你内联设置自定义状态码和响应头。元组中的 body 元素会按相同规则格式化（字符串、字典、字节等）。

注意： 元组必须恰好包含 3 个元素。传入不同长度的元组会引发 ValueError。

py
from robyn import Headers

@app.get("/not-found")
def not_found():
    return (
        {"error": "Resource not found"},
        Headers({"X-Error": "true"}),
        404,
    )
py
from robyn import Headers

@app.get("/not-found")
def not_found():
    return (
        {"error": "Resource not found"},
        Headers({"X-Error": "true"}),
        404,
    )
py
@app.post("/users")
def create_user():
    return (
        {"id": 1, "name": "Alice"},
        Headers({}),
        201,
    )
py
@app.post("/users")
def create_user():
    return (
        {"id": 1, "name": "Alice"},
        Headers({}),
        201,
    )

FileResponse / serve_file / serve_html ​

Robyn 提供用于服务文件的辅助函数。serve_file 会设置 Content-Disposition: attachment 并自动检测 MIME 类型。serve_html 会设置 Content-Type: text/html。你也可以直接构造 FileResponse 以获得完全控制。

py
from robyn import Robyn
from robyn.responses import serve_file

app = Robyn(__file__)

@app.get("/download")
def download():
    return serve_file("report.pdf")
py
from robyn import Robyn
from robyn.responses import serve_file

app = Robyn(__file__)

@app.get("/download")
def download():
    return serve_file("report.pdf")
py
from robyn.responses import serve_html

@app.get("/page")
def page():
    return serve_html("templates/index.html")
py
from robyn.responses import serve_html

@app.get("/page")
def page():
    return serve_html("templates/index.html")
py
from robyn.responses import FileResponse
from robyn.robyn import Headers

@app.get("/custom-file")
def custom_file():
    return FileResponse(
        file_path="data/export.csv",
        status_code=200,
        headers=Headers({"Content-Type": "text/csv"}),
    )
py
from robyn.responses import FileResponse
from robyn.robyn import Headers

@app.get("/custom-file")
def custom_file():
    return FileResponse(
        file_path="data/export.csv",
        status_code=200,
        headers=Headers({"Content-Type": "text/csv"}),
    )

html() ​

html() 辅助将原始 HTML 字符串封装为 Response，并设置 Content-Type: text/html 和状态码 200。适用于在不使用模板引擎的情况下返回动态生成的 HTML。

py
from robyn import Robyn
from robyn.responses import html

app = Robyn(__file__)

@app.get("/page")
def page():
    return html("<h1>Hello, World!</h1><p>Welcome to Robyn.</p>")
py
from robyn import Robyn
from robyn.responses import html

app = Robyn(__file__)

@app.get("/page")
def page():
    return html("<h1>Hello, World!</h1><p>Welcome to Robyn.</p>")

StreamingResponse ​

StreamingResponse 使用生成器发送分块响应（同步或异步）。默认的 media_type 为 text/event-stream，但你可以将其设置为任意 MIME 类型。主体按块流式传输，客户端会在数据生成时逐块接收。

py
from robyn import Robyn
from robyn.responses import StreamingResponse

app = Robyn(__file__)

@app.get("/stream")
def stream():
    def generate():
        for i in range(5):
            yield f"chunk {i}\n"

    return StreamingResponse(
        content=generate(),
        media_type="text/plain",
    )
py
from robyn import Robyn
from robyn.responses import StreamingResponse

app = Robyn(__file__)

@app.get("/stream")
def stream():
    def generate():
        for i in range(5):
            yield f"chunk {i}\n"

    return StreamingResponse(
        content=generate(),
        media_type="text/plain",
    )
py
import asyncio
from robyn.responses import StreamingResponse

@app.get("/stream-async")
async def stream_async():
    async def generate():
        for i in range(5):
            await asyncio.sleep(0.5)
            yield f"chunk {i}\n"

    return StreamingResponse(
        content=generate(),
        media_type="text/plain",
    )
py
import asyncio
from robyn.responses import StreamingResponse

@app.get("/stream-async")
async def stream_async():
    async def generate():
        for i in range(5):
            await asyncio.sleep(0.5)
            yield f"chunk {i}\n"

    return StreamingResponse(
        content=generate(),
        media_type="text/plain",
    )

SSEResponse ​

SSEResponse 是 StreamingResponse 的便捷封装，预配置用于 Server-Sent Events。可配合 SSEMessage 辅助以 SSE 协议格式化消息。每条消息可选包含 event、id、和 retry 字段。

有关 Server-Sent Events 的更详细指南，请参见 SSE 文档。

py
from robyn import Robyn
from robyn.responses import SSEResponse, SSEMessage
import time

app = Robyn(__file__)

@app.get("/events")
def events():
    def event_stream():
        for i in range(10):
            yield SSEMessage(
                f"Event {i}",
                event="update",
                id=str(i),
            )
            time.sleep(1)

    return SSEResponse(event_stream())
py
from robyn import Robyn
from robyn.responses import SSEResponse, SSEMessage
import time

app = Robyn(__file__)

@app.get("/events")
def events():
    def event_stream():
        for i in range(10):
            yield SSEMessage(
                f"Event {i}",
                event="update",
                id=str(i),
            )
            time.sleep(1)

    return SSEResponse(event_stream())
py
import asyncio
from robyn.responses import SSEResponse, SSEMessage

@app.get("/events/async")
async def async_events():
    async def event_stream():
        for i in range(10):
            await asyncio.sleep(0.5)
            yield SSEMessage(
                f"Async event {i}",
                event="update",
                id=str(i),
            )

    return SSEResponse(event_stream())
py
import asyncio
from robyn.responses import SSEResponse, SSEMessage

@app.get("/events/async")
async def async_events():
    async def event_stream():
        for i in range(10):
            await asyncio.sleep(0.5)
            yield SSEMessage(
                f"Async event {i}",
                event="update",
                id=str(i),
            )

    return SSEResponse(event_stream())

接下来？ ​

现在你已经了解了 Robyn 处理程序返回数据的各种方式，可以继续学习文件上传以了解如何从客户端接收文件。

基于 MIT 许可发布

版权所有 © 2025-2026 王则钰