FastAPI自学教程（60） - 直接使用请求对象

1.基础用法

from fastapi import FastAPI, Request

app = FastAPI()

@app.get("/items/{item_id}")
def read_root(item_id: str, request: Request):
    client_host = request.client.host
    return {"client_host": client_host, "item_id": item_id}

• 通过声明Request类型参数直接获取请求对象
• 请求处理流程：

  graph TD
    A[请求到达] --> B{路径参数解析}
    B --> C[注入Request对象]
    C --> D[访问client.host等属性]
    D --> E[返回响应数据]

2.请求对象核心特性

2.1 数据访问层级

• 基础属性：

  request.method        # 请求方法（GET/POST等）
  request.url.path      # 请求路径
  request.client.port   # 客户端端口号

• 底层数据：

  request.scope        # 原始ASGI scope字典
  request.scope["headers"]  # 原始头信息字节列表

• 协议信息：

  request.base_url     # 基础URL（含协议和主机）
  request.http_version  # HTTP协议版本

2.2 数据获取方法

# 获取查询参数
search_term = request.query_params.get("q")

# 获取路径参数
item_id = request.path_params["item_id"]

# 获取请求头
user_agent = request.headers.get("user-agent")

# 获取cookie
session_id = request.cookies.get("session")

3.进阶配置

3.1 中间件集成

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

• 在中间件中访问请求对象实现全局处理

3.2 数据流处理

@app.post("/upload")
async def upload_file(request: Request):
    body = b""
    async for chunk in request.stream():
        body += chunk
    return {"file_size": len(body)}

• 使用流式处理应对大文件上传场景

4.生产环境建议

1. 安全访问：

   • 使用request.client代替直接解析X-Forwarded-For头
   • 敏感信息访问需配合HTTPS加密

2. 性能优化：

   • 避免在业务逻辑中频繁访问request.scope
   • 对常用数据做本地缓存（如request.method）

3. 错误处理：

   from starlette.exceptions import HTTPException
   @app.exception_handler(HTTPException)
   async def custom_handler(request: Request, exc: HTTPException):
       return JSONResponse({"detail": exc.detail}, status_code=exc.status_code)

测试命令示例：

curl http://localhost:8000/items/123