高级路由与参数注入

Robyn 的路由系统远不止简单的 URL 匹配。它包含复杂的参数注入、路由优化和灵活的模式匹配，使构建复杂 API 变得轻而易举。

理解参数注入

Robyn 会自动分析你的函数签名并注入合适的请求组件。这消除了样板代码并使处理器更为简洁。

注入引擎

参数注入系统分为两个阶段：

函数内省: Robyn 在注册时分析你的函数签名
运行时注入: 对于每个请求，Robyn 提供函数所需的确切参数

基于类型的注入: 使用类型注解来确定要注入的内容

Type-Based Parameter Injection

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

@app.post("/users/:user_id/posts/:post_id")
async def update_post(
    # Robyn automatically injects these based on type annotations
    request: Request,           # Complete request object
    path_params: PathParams,    # {"user_id": "123", "post_id": "456"}
    query_params: QueryParams,  # ?draft=true&tags=python,web
    headers: Headers,           # All request headers
    body: RequestBody,         # Raw request body
    method: RequestMethod      # "POST"
):
    user_id = path_params["user_id"]
    post_id = path_params["post_id"]
    is_draft = query_params.get("draft") == "true"
    content_type = headers.get("content-type")
    
    return {
        "user_id": user_id,
        "post_id": post_id,
        "is_draft": is_draft,
        "body_size": len(body),
        "method": method
    }

基于名称的注入: 当没有类型注解时，使用参数名称注入组件

Name-Based Parameter Injection

# Reserved parameter names that Robyn recognizes
@app.get("/search/:category")
def search_handler(
    query_params,      # Injected by name
    path_params,       # Injected by name
    headers,           # Injected by name
    request           # Injected by name (full request object)
):
    category = path_params["category"]
    search_term = query_params.get("q", "")
    user_agent = headers.get("user-agent", "")
    
    return {
        "category": category,
        "search": search_term,
        "user_agent": user_agent
    }

可注入类型完整列表

类型注解	保留名称	描述
`Request`	`request`, `req`, `r`	完整的请求对象
`QueryParams`	`query_params`	URL 查询参数
`Headers`	`headers`	请求头
`PathParams`	`path_params`	URL 路径参数
`RequestBody`	`body`	原始请求体
`RequestMethod`	`method`	HTTP 方法（GET、POST 等）
`RequestURL`	`url`	请求 URL 信息
`FormData`	`form_data`	表单编码的数据
`RequestFiles`	`files`	上传的文件
`RequestIP`	`ip_addr`	客户端 IP 地址
`RequestIdentity`	`identity`	认证身份

高级 URL 模式

动态路由参数

Robyn 支持多种类型的路径参数及灵活的匹配模式。

Dynamic Route Patterns

# Simple parameter
@app.get("/users/:id")
def get_user(path_params):
    return {"user_id": path_params["id"]}

# Multiple parameters
@app.get("/users/:user_id/posts/:post_id")
def get_user_post(path_params):
    return {
        "user_id": path_params["user_id"],
        "post_id": path_params["post_id"]
    }

# Optional parameters with defaults
@app.get("/posts/:id/:slug?")
def get_post(path_params):
    post_id = path_params["id"]
    slug = path_params.get("slug", f"post-{post_id}")
    return {"id": post_id, "slug": slug}

# Wildcard matching
@app.get("/files/*filepath")
def serve_file(path_params):
    filepath = path_params["filepath"]
    return {"serving": filepath}

路由约束与验证

虽然 Robyn 本身没有内置参数验证，但你可以在处理器中实现以保证类型安全。

Parameter Validation

import re
from robyn import HTTPException

@app.get("/users/:user_id")
def get_user(path_params):
    user_id = path_params["user_id"]
    
    # Validate that user_id is numeric
    if not user_id.isdigit():
        raise HTTPException(400, "user_id must be numeric")
    
    user_id = int(user_id)
    if user_id <= 0:
        raise HTTPException(400, "user_id must be positive")
    
    return {"user_id": user_id}

@app.get("/posts/:slug")
def get_post_by_slug(path_params):
    slug = path_params["slug"]
    
    # Validate slug format
    if not re.match(r'^[a-z0-9-]+$', slug):
        raise HTTPException(400, "Invalid slug format")
    
    return {"slug": slug}

路由优化

用于静态响应的 Const 路由

对于永不改变的响应使用 const=True。这些响应会被缓存在 Rust 内存中，处理器函数在启动后不会再次执行。

如果没有注册任何中间件，const 路由会走完全由 Rust 层处理的快速路径而不进入 Python。当注册了中间件（包括全局的 before-request 与 after-request）时，const 路由仍然会返回缓存响应，但中间件会在每次请求时正常执行。这意味着 const 路由可以安全地与中间件一起使用。

Const Route Optimization

# Perfect for health checks, static configuration
@app.get("/health", const=True)
def health_check():
    return {"status": "healthy", "version": "1.0"}

# API metadata that rarely changes
@app.get("/api/info", const=True)  
def api_info():
    return {
        "name": "My API",
        "version": "2.1.0",
        "documentation": "/docs"
    }

# Static configuration endpoints
@app.get("/config/public", const=True)
def public_config():
    return {
        "max_upload_size": "10MB",
        "allowed_origins": ["https://myapp.com"]
    }

路由优先级与顺序

路由按注册顺序匹配。更具体的路由应在通用路由之前注册。

Route Ordering Best Practices

# GOOD: Specific routes first
@app.get("/users/profile")
def get_current_user_profile():
    return {"profile": "current_user"}

@app.get("/users/settings")
def get_user_settings():
    return {"settings": "user_settings"}

@app.get("/users/:id")
def get_user(path_params):
    return {"user_id": path_params["id"]}

# BAD: This would never be reached
# @app.get("/users/:id")  # Registered first
# @app.get("/users/profile")  # Never matched!

高级查询参数处理

查询参数解析

Robyn 提供丰富的查询参数处理，并带有自动类型转换的辅助方法。

Advanced Query Parameters

@app.get("/search")
def search(query_params):
    # Basic parameter access
    q = query_params.get("q", "")
    
    # Parameters with defaults
    page = int(query_params.get("page", "1"))
    limit = int(query_params.get("limit", "10"))
    
    # Boolean parameters
    include_deleted = query_params.get("include_deleted", "false").lower() == "true"
    
    # Array parameters (?tags=python&tags=web&tags=api)
    tags = query_params.get_list("tags") or []
    
    # Convert to dict for easier processing
    all_params = query_params.to_dict()
    
    return {
        "query": q,
        "page": page,
        "limit": limit,
        "include_deleted": include_deleted,
        "tags": tags,
        "all_params": all_params
    }

复杂查询字符串模式

处理如过滤、排序和嵌套参数等复杂查询模式。

Complex Query Patterns

@app.get("/api/products")
def get_products(query_params):
    # Filtering: ?filter[category]=electronics&filter[price_min]=100
    filters = {}
    for key, value in query_params.to_dict().items():
        if key.startswith("filter[") and key.endswith("]"):
            filter_key = key[7:-1]  # Remove "filter[" and "]"
            filters[filter_key] = value
    
    # Sorting: ?sort=price&order=desc
    sort_field = query_params.get("sort", "created_at")
    sort_order = query_params.get("order", "asc")
    
    # Pagination: ?page=2&per_page=20
    page = int(query_params.get("page", "1"))
    per_page = min(int(query_params.get("per_page", "10")), 100)  # Cap at 100
    
    # Field selection: ?fields=id,name,price
    fields = query_params.get("fields", "").split(",") if query_params.get("fields") else None
    
    return {
        "filters": filters,
        "sort": {"field": sort_field, "order": sort_order},
        "pagination": {"page": page, "per_page": per_page},
        "fields": fields
    }

子路由与模块化路由

创建子路由器

子路由器帮助通过为相关路由分组并使用公共前缀来组织大型应用。

SubRouter Organization

from robyn import Robyn, SubRouter

app = Robyn(__file__)

# API v1 routes
api_v1 = SubRouter(__file__, prefix="/api/v1")

@api_v1.get("/users")
def list_users():
    return {"users": []}

@api_v1.get("/users/:id")
def get_user(path_params):
    return {"user_id": path_params["id"]}

@api_v1.post("/users")
def create_user(body):
    return {"created": True, "data": body}

# Admin routes
admin = SubRouter(__file__, prefix="/admin")

@admin.get("/dashboard")
def admin_dashboard():
    return {"dashboard": "admin"}

@admin.get("/users")
def admin_users():
    return {"admin_users": []}

# Register subrouters
app.include_router(api_v1)
app.include_router(admin)

# Routes are now available at:
# /api/v1/users, /api/v1/users/:id, /admin/dashboard, etc.

子路由器中间件

在子路由器上配置认证处理器，并使用 auth_required=True 将认证应用于路由。

SubRouter Middleware

from robyn import SubRouter
from robyn.authentication import AuthenticationHandler

# Admin routes with authentication
admin = SubRouter(__file__, prefix="/admin")

class AdminAuth(AuthenticationHandler):
    def authenticate(self, request):
        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            return None
        
        token = auth_header[7:]  # Remove "Bearer "
        return self.validate_admin_token(token)
    
    def validate_admin_token(self, token):
        # Your token validation logic
        if token == "admin-secret-token":
            return {"user": "admin"}  # Return identity object
        return None

# Configure the authentication handler for this SubRouter
admin.configure_authentication(AdminAuth())

# Routes must explicitly require authentication with auth_required=True
@admin.get("/users", auth_required=True)
def admin_users():
    return {"admin_users": ["user1", "user2"]}

@admin.delete("/users/:id", auth_required=True)
def delete_user(path_params):
    return {"deleted": path_params["id"]}

路由测试与调试

路由检查

通过检查已注册的路由表来调试你的路由。

Route Debugging

from robyn import Robyn

app = Robyn(__file__)

@app.get("/users/:id")
def get_user(path_params):
    return {"user": path_params["id"]}

@app.post("/users")
def create_user(body):
    return {"created": True}

# Debug: Print all registered routes
if __name__ == "__main__":
    print("Registered routes:")
    for route in app.get_routes():
        print(f"{route.method} {route.path}")
    
    app.start(port=8080)

路由性能监控

添加计时中间件来监控路由性能。

Performance Monitoring

import time
from robyn import Robyn

app = Robyn(__file__)

@app.before_request
def timing_middleware(request):
    request.start_time = time.time()
    return request

@app.after_request
def timing_after_middleware(request, response):
    duration = time.time() - request.start_time
    print(f"{request.method} {request.url.path} - {duration:.3f}s")
    response.headers["X-Response-Time"] = f"{duration:.3f}s"
    return response

@app.get("/slow")
def slow_endpoint():
    time.sleep(0.1)  # Simulate work
    return {"message": "slow response"}

最佳实践

1. 参数注入模式

使用类型注解 以获得更好的 IDE 支持和自文档化代码
只注入所需内容 以保持处理器专注
尽早验证参数 以提供清晰的错误信息

2. 路由组织

分组相关路由 使用子路由器
将路由按从具体到通用的顺序排列 以避免匹配问题
使用一致的命名约定 对路径参数

3. 性能优化

使用 const 路由 对静态响应
减少参数注入 在高流量端点中
缓存昂贵的计算 而不是重复执行

4. 错误处理

尽早验证路径参数
为无效输入提供有意义的错误信息
在整个 API 中使用一致的错误响应格式

接下来可以做什么？

现在你已经掌握了高级路由，继续探索其他 Robyn 功能：