4. FastAPI 核心概念与第一个接口：路由与路径操作装饰器、请求与响应模型（Pydantic）、启动服务与 Swagger 文档自动生成

本章是构建信用评分特征服务的起点。我们将从零搭建一个 FastAPI 项目，理解其最核心的三个概念：路由与路径操作装饰器、Pydantic 请求/响应模型，以及自动生成的 Swagger 交互文档。掌握这些基础后，你就能快速定义出第一个可用于信用评分特征查询的 API 端点。

4.1 路由与路径操作装饰器

FastAPI 使用 Python 装饰器将函数绑定到特定的 HTTP 方法和 URL 路径上。这种设计让代码声明式、可读性极强。

• 路由（Route）：客户端访问的 URL 路径，例如 /features/credit_score。
• 路径操作装饰器（Path Operation Decorator）：如 @app.get()、@app.post()，用于声明 HTTP 方法。
• 路径操作函数（Path Operation Function）：被装饰器包裹的 Python 函数，处理请求并返回响应。

下表总结了信用评分特征服务中最常用的 HTTP 方法与对应场景：

以下代码展示了如何定义一个最简单的 GET 接口：

from fastapi import FastAPI

app = FastAPI(title="信用评分特征服务")

@app.get("/")
def read_root():
    return {"message": "信用评分特征服务已启动"}

@app.get("/features/credit_score/{user_id}")
def get_credit_score(user_id: int):
    # 模拟查询数据库
    return {"user_id": user_id, "credit_score": 750, "risk_level": "低风险"}

4.2 请求与响应模型（Pydantic）

Pydantic 是 FastAPI 的数据验证核心。通过定义数据模型，你可以自动完成请求体的解析、类型校验和响应格式的规范化。在信用评分场景中，这能确保传入的特征数据格式正确，避免脏数据进入评分引擎。

• 请求模型：定义客户端发送的 JSON 数据结构，通常用于 POST/PUT 请求。
• 响应模型：定义 API 返回给客户端的数据结构，确保输出一致且安全（可隐藏敏感字段）。
• 自动验证：Pydantic 会在运行时校验字段类型、范围、必填性等，校验失败时自动返回 422 错误。

下面是一个完整的 Pydantic 模型定义示例：

from pydantic import BaseModel, Field
from typing import Optional
from datetime import date

# 请求模型：用户特征提交
class UserFeaturesRequest(BaseModel):
    user_id: int = Field(..., description="用户唯一标识")
    age: int = Field(..., ge=18, le=120, description="年龄（18-120）")
    annual_income: float = Field(..., gt=0, description="年收入（元）")
    existing_loans: int = Field(0, ge=0, description="现有贷款笔数")
    credit_history_years: float = Field(..., ge=0, description="信用历史年限")
    application_date: date = Field(..., description="申请日期")

# 响应模型：评分结果
class CreditScoreResponse(BaseModel):
    user_id: int
    credit_score: int = Field(..., ge=300, le=850, description="信用评分（300-850）")
    risk_level: str = Field(..., description="风险等级")
    is_approved: bool = Field(..., description="是否批准")
    processing_time_ms: Optional[float] = Field(None, description="处理耗时（毫秒）")

在接口中使用这些模型：

from fastapi import FastAPI, HTTPException

app = FastAPI(title="信用评分特征服务")

@app.post("/features/submit", response_model=CreditScoreResponse)
def submit_features(features: UserFeaturesRequest):
    # 模拟评分计算逻辑
    score = 650 + int(features.annual_income / 10000) - features.existing_loans * 10
    score = max(300, min(850, score))
    
    risk = "低风险" if score >= 700 else ("中风险" if score >= 600 else "高风险")
    approved = score >= 600
    
    return CreditScoreResponse(
        user_id=features.user_id,
        credit_score=score,
        risk_level=risk,
        is_approved=approved,
        processing_time_ms=12.5
    )

4.3 启动服务与 Swagger 文档自动生成

FastAPI 内置了自动生成 OpenAPI 规范文档的能力，并提供了两个开箱即用的交互式文档界面：Swagger UI（/docs）和 ReDoc（/redoc）。你无需编写任何文档代码，只需定义好路由和 Pydantic 模型，文档就会自动生成。

启动服务的标准方式：

# 终端命令
uvicorn main:app --reload --host 0.0.0.0 --port 8000

• main:app：main.py 文件中的 app 实例。
• --reload：开发模式下自动重载，代码变更后自动重启。
• --host 0.0.0.0：允许外部访问（默认仅本地）。
• --port 8000：监听端口。

启动后访问：

• Swagger UI：http://localhost:8000/docs —— 可在线测试每个接口。
• ReDoc：http://localhost:8000/redoc —— 更清晰的文档阅读视图。

4.4 核心流程总览

下面的 SVG 图展示了 FastAPI 从接收请求到返回响应的完整流程，以及路由、Pydantic 模型和自动文档之间的关系：

4.5 完整示例：第一个信用评分特征接口

将以上所有概念整合到一个完整的 main.py 文件中：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import Optional
from datetime import date
import time

# ---------- 1. 创建应用实例 ----------
app = FastAPI(
    title="信用评分特征服务",
    description="提供用户信用特征提交与评分查询的 API 服务",
    version="1.0.0"
)

# ---------- 2. 定义 Pydantic 模型 ----------
class UserFeaturesRequest(BaseModel):
    user_id: int = Field(..., description="用户唯一标识")
    age: int = Field(..., ge=18, le=120, description="年龄")
    annual_income: float = Field(..., gt=0, description="年收入（元）")
    existing_loans: int = Field(0, ge=0, description="现有贷款笔数")
    credit_history_years: float = Field(..., ge=0, description="信用历史年限")

class CreditScoreResponse(BaseModel):
    user_id: int
    credit_score: int = Field(..., ge=300, le=850, description="信用评分")
    risk_level: str = Field(..., description="风险等级")
    is_approved: bool = Field(..., description="是否批准")
    processing_time_ms: float = Field(..., description="处理耗时（毫秒）")

# ---------- 3. 定义路由与路径操作函数 ----------
@app.get("/")
def root():
    """根路径，返回服务状态"""
    return {"service": "信用评分特征服务", "status": "running"}

@app.get("/features/credit_score/{user_id}")
def get_credit_score(user_id: int):
    """根据用户 ID 查询信用评分（模拟数据）"""
    # 模拟数据库查询
    mock_scores = {1001: 780, 1002: 620, 1003: 550}
    score = mock_scores.get(user_id)
    if score is None:
        raise HTTPException(status_code=404, detail="用户不存在")
    
    risk = "低风险" if score >= 700 else ("中风险" if score >= 600 else "高风险")
    return {"user_id": user_id, "credit_score": score, "risk_level": risk}

@app.post("/features/submit", response_model=CreditScoreResponse)
def submit_features(features: UserFeaturesRequest):
    """提交用户特征数据并返回评分结果"""
    start_time = time.time()
    
    # 模拟评分计算逻辑
    base_score = 650
    income_bonus = int(features.annual_income / 10000) * 5
    loan_penalty = features.existing_loans * 15
    history_bonus = int(features.credit_history_years) * 3
    age_factor = 10 if 25 <= features.age <= 55 else -10
    
    score = base_score + income_bonus - loan_penalty + history_bonus + age_factor
    score = max(300, min(850, score))
    
    risk = "低风险" if score >= 700 else ("中风险" if score >= 600 else "高风险")
    approved = score >= 600
    
    elapsed = (time.time() - start_time) * 1000
    
    return CreditScoreResponse(
        user_id=features.user_id,
        credit_score=score,
        risk_level=risk,
        is_approved=approved,
        processing_time_ms=round(elapsed, 2)
    )

启动服务后，访问 http://localhost:8000/docs 即可看到自动生成的 Swagger 文档，并可直接在页面中测试 POST /features/submit 接口。

4.6 本章小结