催收API熔断机制从入门到实战 - 第4节

4.1 Flask 环境搭建与项目初始化

Flask 是 Python 生态中最轻量的 Web 框架之一，非常适合快速搭建催收 API 原型。本节将完成 Flask 安装、虚拟环境配置和项目骨架创建。

4.1.1 安装 Flask

推荐在虚拟环境中安装 Flask，避免依赖冲突：

# 创建虚拟环境（Python 3.8+）
python -m venv venv_collection

# 激活虚拟环境（Windows）
venv_collection\Scripts\activate

# 激活虚拟环境（macOS/Linux）
source venv_collection/bin/activate

# 安装 Flask
pip install flask

# 验证安装
python -c "import flask; print(flask.__version__)"

4.1.2 项目目录结构

一个标准的催收 API 项目目录结构如下：

collection_api/
├── app.py                  # 应用入口
├── config.py               # 配置文件
├── requirements.txt        # 依赖清单
├── routes/
│   ├── __init__.py
│   └── collection.py       # 催收路由模块
├── models/
│   ├── __init__.py
│   └── task.py             # 催收任务模型
└── utils/
    ├── __init__.py
    └── response.py         # 统一响应工具

4.2 创建第一个催收任务查询接口

我们从最简单的单文件应用开始，逐步拆解 Flask 路由的核心概念。

4.2.1 最小化 Flask 应用

在 app.py 中编写以下代码：

from flask import Flask, jsonify, request

app = Flask(__name__)

# 模拟催收任务数据
MOCK_TASKS = {
    "TASK001": {
        "task_id": "TASK001",
        "debtor_name": "张三",
        "amount": 15000.00,
        "status": "进行中",
        "created_at": "2025-03-01 10:30:00"
    },
    "TASK002": {
        "task_id": "TASK002",
        "debtor_name": "李四",
        "amount": 28000.50,
        "status": "待分配",
        "created_at": "2025-03-02 14:15:00"
    }
}

@app.route('/api/v1/tasks/<task_id>', methods=['GET'])
def get_task(task_id):
    """根据任务ID查询催收任务详情"""
    task = MOCK_TASKS.get(task_id)
    if not task:
        return jsonify({
            "code": 404,
            "message": f"任务 {task_id} 不存在",
            "data": None
        }), 404
    return jsonify({
        "code": 200,
        "message": "查询成功",
        "data": task
    }), 200

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, debug=True)

4.2.2 启动与测试

在终端执行：

python app.py

使用 curl 测试接口：

# 查询存在的任务
curl http://127.0.0.1:5000/api/v1/tasks/TASK001

# 查询不存在的任务
curl http://127.0.0.1:5000/api/v1/tasks/TASK999

{
  "code": 200,
  "message": "查询成功",
  "data": {
    "task_id": "TASK001",
    "debtor_name": "张三",
    "amount": 15000.00,
    "status": "进行中",
    "created_at": "2025-03-01 10:30:00"
  }
}

4.3 路由与请求方法

Flask 路由系统通过 @app.route() 装饰器将 URL 规则绑定到视图函数。催收 API 中常见的路由模式如下：

4.3.1 路由参数与类型转换

Flask 支持在路由中指定参数类型，避免手动类型转换：

# 字符串参数（默认）
@app.route('/api/v1/tasks/<task_id>')

# 整数参数
@app.route('/api/v1/collectors/<int:collector_id>')

# 浮点数参数
@app.route('/api/v1/amount/<float:min_amount>')

# UUID 参数
@app.route('/api/v1/records/<uuid:record_id>')

4.4 JSON 响应格式设计

统一的 JSON 响应格式是 API 可维护性的基石。催收系统需要在前端、催收员 App、第三方渠道之间保持一致的响应契约。

4.4.1 标准响应结构

我们定义三层响应结构：

{
  "code": 200,          // 业务状态码，非 HTTP 状态码
  "message": "成功",    // 人类可读的消息
  "data": {             // 实际业务数据
    "task_id": "TASK001",
    "debtor_name": "张三",
    ...
  },
  "timestamp": 1710000000,  // 响应时间戳（可选）
  "request_id": "req_abc"   // 请求追踪 ID（可选）
}

4.4.2 业务状态码定义

4.4.3 统一响应工具函数

创建 utils/response.py 封装响应生成逻辑：

from flask import jsonify
import time
import uuid

def success(data=None, message="成功", code=200):
    """生成成功响应"""
    return jsonify({
        "code": code,
        "message": message,
        "data": data,
        "timestamp": int(time.time()),
        "request_id": str(uuid.uuid4())[:8]
    }), 200

def error(message="服务异常", code=500, http_status=500):
    """生成错误响应"""
    return jsonify({
        "code": code,
        "message": message,
        "data": None,
        "timestamp": int(time.time()),
        "request_id": str(uuid.uuid4())[:8]
    }), http_status

使用工具函数重构查询接口：

from utils.response import success, error

@app.route('/api/v1/tasks/<task_id>', methods=['GET'])
def get_task(task_id):
    task = MOCK_TASKS.get(task_id)
    if not task:
        return error(f"任务 {task_id} 不存在", code=404, http_status=404)
    return success(data=task)

4.5 催收 API 请求处理流程

下图展示了从客户端发起请求到返回 JSON 响应的完整链路：

4.6 本章小结

• 环境搭建：使用虚拟环境隔离依赖，Flask 2.x 稳定可靠。
• 路由设计：催收 API 遵循 RESTful 风格，路径以 /api/v1/ 开头，支持版本管理。
• 请求方法：GET 用于查询，POST 用于创建，PUT 用于更新，DELETE 用于删除。
• JSON 响应：统一使用 code + message + data 三层结构，业务状态码与 HTTP 状态码解耦。
• 工具函数：封装 success() 和 error() 函数，避免重复编写响应逻辑。