4. FastAPI 框架入门：核心特性、项目结构与数据校验

本章是构建信用评分 API 的技术起点。FastAPI 凭借异步原生、自动生成 OpenAPI 文档和基于 Pydantic 的强类型校验，成为金融级低延迟接口的首选框架。我们将从核心特性、项目骨架搭建到数据模型定义，完成第一个可运行的评分端点。

4.1 FastAPI 核心特性

FastAPI 的设计围绕三个关键能力展开，直接决定了信用评分 API 的开发效率与运行时性能。

4.2 项目结构搭建

一个可维护的信用评分 API 项目需要清晰的目录分层。我们采用模块化结构，将路由、模型、服务、配置分离，便于团队协作与单元测试。

credit_score_api/
├── app/
│   ├── __init__.py
│   ├── main.py              # FastAPI 应用入口
│   ├── config.py            # 配置管理（数据库、缓存、密钥）
│   ├── routers/
│   │   ├── __init__.py
│   │   └── score.py         # 评分相关路由
│   ├── models/
│   │   ├── __init__.py
│   │   └── schemas.py       # Pydantic 请求/响应模型
│   ├── services/
│   │   ├── __init__.py
│   │   └── scoring.py       # 评分业务逻辑
│   └── dependencies.py      # 依赖注入（数据库会话、认证）
├── tests/
│   ├── __init__.py
│   └── test_score.py
├── requirements.txt
└── Dockerfile

在 main.py 中创建应用实例并注册路由：

from fastapi import FastAPI
from app.routers import score

app = FastAPI(
    title="信用评分 API",
    description="低延迟信用评分接口，支持异步批量评分",
    version="1.0.0"
)

app.include_router(score.router, prefix="/api/v1", tags=["评分"])

4.3 Pydantic 模型定义与数据校验

Pydantic 是 FastAPI 的数据校验基石。通过 Python 类型注解，我们声明请求体、响应体和查询参数的结构，FastAPI 自动完成解析、校验和文档生成。

4.3.1 基础模型定义

以下是一个信用评分请求的 Pydantic 模型：

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

class CreditScoreRequest(BaseModel):
    user_id: str = Field(..., description="用户唯一标识")
    income: float = Field(..., gt=0, description="年收入（元）")
    age: int = Field(..., ge=18, le=100, description="年龄")
    existing_loans: int = Field(0, ge=0, description="现有贷款笔数")
    credit_history_months: int = Field(..., ge=0, description="信用历史月数")
    request_date: Optional[date] = Field(None, description="请求日期")

    @field_validator("income")
    @classmethod
    def income_must_be_reasonable(cls, v: float) -> float:
        if v > 1_000_000_000:
            raise ValueError("收入金额异常，请核实")
        return v

4.3.2 响应模型与嵌套模型

响应模型同样使用 Pydantic 定义，确保返回数据格式一致：

class CreditScoreResponse(BaseModel):
    user_id: str
    score: int = Field(..., ge=300, le=850, description="信用评分")
    level: str = Field(..., description="信用等级：A/B/C/D")
    processing_time_ms: float = Field(..., description="处理耗时（毫秒）")

class BatchScoreRequest(BaseModel):
    requests: list[CreditScoreRequest] = Field(..., max_length=100)

class BatchScoreResponse(BaseModel):
    results: list[CreditScoreResponse]
    total_processing_time_ms: float

4.3.3 在路由中使用模型

FastAPI 自动将请求体解析为 Pydantic 模型实例，并执行所有校验：

from fastapi import APIRouter, Depends
from app.models.schemas import CreditScoreRequest, CreditScoreResponse
from app.services.scoring import calculate_score

router = APIRouter()

@router.post("/score", response_model=CreditScoreResponse)
async def get_credit_score(request: CreditScoreRequest):
    """
    获取单个用户的信用评分
    """
    score, level, elapsed = await calculate_score(request)
    return CreditScoreResponse(
        user_id=request.user_id,
        score=score,
        level=level,
        processing_time_ms=elapsed
    )

4.4 启动与验证

完成模型和路由定义后，通过以下命令启动服务：

uvicorn app.main:app --reload --host 0.0.0.0 --port 8000

访问 http://localhost:8000/docs 即可看到自动生成的 Swagger UI 文档，所有请求/响应模型、校验规则、示例值一目了然。发送一个测试请求验证校验效果：

curl -X POST "http://localhost:8000/api/v1/score" \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "U001",
    "income": 500000,
    "age": 30,
    "existing_loans": 2,
    "credit_history_months": 60
  }'