Python：做接口工具和数据处理脚本时，代码怎样写才不会越写越乱

发表于 分类于 本文字数： 3.4k 阅读时长 ≈ 12 分钟
Python 很适合写接口工具和数据处理脚本，但也最容易从一个文件长成一团。这篇文章围绕一个接口巡检和结果清洗脚本，讲清楚代码怎样拆、配置怎样放、输出怎样组织，才不会越写越乱。

Python 做接口工具和数据处理脚本时，上手通常很快，失控也通常很快。

最常见的演变过程一般是这样：

  • 第一版只有几十行
  • 后面加了请求头、鉴权、重试
  • 再后面加了结果输出、告警、数据清洗
  • 再后来又要兼容不同环境和不同接口

代码就会慢慢长成下面这种状态：

  • 所有逻辑都挤在一个文件里
  • 配置、请求、清洗、输出混在一起
  • 修改一个字段，三四个地方都要跟着改
  • 一出错就只能靠打印排查
  • 加一个新接口比重写半个脚本还累

这一篇不讲空泛原则，直接围绕一个实际功能展开：做一个接口巡检和结果清洗脚本。

这个脚本要完成这些事：

  1. 调用一组接口
  2. 校验响应状态
  3. 提取几个关键字段
  4. 清洗成统一结果
  5. 导出成 JSON 文件

用这个场景把最容易越写越乱的几块拆开：

  1. 请求逻辑和业务校验为什么不能揉在一起
  2. 配置、接口定义、输出模型怎么拆
  3. 数据清洗为什么最好单独成函数
  4. 一份脚本什么时候该从单文件走向目录结构

一、这篇文章要解决什么问题

读完这一篇，应该能独立完成这些动作：

  • 把一个接口脚本拆成至少三层职责
  • 把配置和执行逻辑分开
  • 把响应清洗成统一结构
  • 判断哪些逻辑适合抽成函数，哪些该留在主流程
  • 识别几类典型坏味道，例如参数满天飞、函数职责混杂、输出结构不一致

如果这些动作能独立做出来，Python 脚本就不再只是“能跑一遍”，而是开始能长期维护。

二、先看要完成的实际功能

假设现在要做一个最简单的接口巡检工具，检查三类接口：

  • 登录状态接口
  • 订单健康接口
  • 配置中心接口

目标输出类似这样：

1
2
3
4
5
6
巡检完成，总接口数: 3
成功: 2
失败: 1

失败接口:
- config-service: timeout

同时把结果导出成 report.json：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
[
  {
    "service": "auth-service",
    "ok": true,
    "message": "ok"
  },
  {
    "service": "order-service",
    "ok": true,
    "message": "ok"
  },
  {
    "service": "config-service",
    "ok": false,
    "message": "timeout"
  }
]

这个需求很小，但已经足够把“工具代码怎么避免失控”讲透。

三、先看最容易失控的版本

很多脚本的第一版通常会长这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import json
import requests


token = "demo-token"
result = []

resp1 = requests.get("https://example.com/auth/status", headers={"Authorization": token}, timeout=3)
if resp1.status_code == 200 and resp1.json()["code"] == 0:
    result.append({"service": "auth-service", "ok": True, "message": "ok"})
else:
    result.append({"service": "auth-service", "ok": False, "message": "error"})

resp2 = requests.get("https://example.com/order/health", headers={"Authorization": token}, timeout=3)
if resp2.status_code == 200 and resp2.json()["status"] == "UP":
    result.append({"service": "order-service", "ok": True, "message": "ok"})
else:
    result.append({"service": "order-service", "ok": False, "message": "error"})

print(json.dumps(result, ensure_ascii=False, indent=2))

这段代码短期能跑，但问题已经很明显：

  • 请求构造重复
  • 判断逻辑分散
  • 输出结构靠手写拼
  • 服务一多就会开始复制粘贴

这不是代码“写得丑”，而是职责还没拆开。

四、先拆第一层：把接口定义和执行分开

1. 接口定义先集中起来

先不要每个地方都写 URL 和名字，把接口定义收拢：

1
2
3
4
5
ENDPOINTS = [
    {"service": "auth-service", "url": "https://example.com/auth/status"},
    {"service": "order-service", "url": "https://example.com/order/health"},
    {"service": "config-service", "url": "https://example.com/config/ping"},
]

这样做最直接的好处是：

  • 新增接口时先改一处
  • 扫一眼就知道脚本到底在检查哪些服务

2. 请求逻辑单独抽成函数

1
2
3
4
5
def request_json(url, token, timeout=3):
    headers = {"Authorization": token}
    response = requests.get(url, headers=headers, timeout=timeout)
    response.raise_for_status()
    return response.json()

这里先不谈高级封装，先解决最实际的问题：不要每个接口都手写同样的请求动作。

3. 一个实际错误：请求逻辑和业务判断缠在一起

错误写法通常像这样：

1
2
3
response = requests.get(url, headers=headers, timeout=3)
if response.status_code == 200 and response.json()["code"] == 0:
    ...

这里的问题是：

  • HTTP 成功判断
  • JSON 解析
  • 业务字段校验

三层逻辑全揉在一起了。后面一旦某个接口返回结构不同，整个主流程就会越来越难看。

五、再拆第二层：把“怎么判断成功”单独写出来

不同接口的成功条件通常不一样：

  • 有的看 code == 0
  • 有的看 status == "UP"
  • 有的看 success is True

如果这层不拆出来，主流程很快就会变成一堆 if/elif/else。

1. 先写不同接口的校验函数

1
2
3
4
5
6
7
8
9
10
def check_auth_service(data):
    return data.get("code") == 0


def check_order_service(data):
    return data.get("status") == "UP"


def check_config_service(data):
    return data.get("success") is True

2. 再建立服务到校验函数的映射

1
2
3
4
5
CHECKERS = {
    "auth-service": check_auth_service,
    "order-service": check_order_service,
    "config-service": check_config_service,
}

这样后面主流程只需要：

1
2
checker = CHECKERS[service_name]
ok = checker(data)

3. 一个实际错误：所有接口共用同一套判断

错误写法：

1
ok = data.get("code") == 0

如果所有接口都这么判断，order-service 一类的接口就会直接误判。

所以“统一执行流程”和“统一判断规则”不是一回事。执行流程可以统一，判断规则通常要按服务拆开。

六、再拆第三层：输出结构必须统一

很多脚本后面难以继续维护，不是因为请求不会写，而是因为输出结构前后不一致。

例如有人第一条结果这样写：

1
{"service": "auth-service", "ok": True, "message": "ok"}

第二条结果又写成：

1
{"name": "order-service", "success": True, "msg": "ok"}

这会直接导致：

  • 后续汇总逻辑要写很多兼容代码
  • 导出 JSON 时字段不统一
  • 排查失败时很难一次性看懂

1. 先写统一的结果清洗函数

1
2
3
4
5
6
def build_result(service, ok, message):
    return {
        "service": service,
        "ok": ok,
        "message": message,
    }

2. 一个实际错误：主流程直接手写结果字典

错误写法：

1
2
3
4
5
results.append({
    "name": service_name,
    "success": True,
    "msg": "ok",
})

修复写法：

1
results.append(build_result(service_name, True, "ok"))

如果统一结果结构这一步不做，后面加报表、加告警、加归档都会开始痛苦。

七、把脚本真正整理成可维护版本

现在把前面的拆分真正串起来。

api_probe.py：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
import json
import requests


ENDPOINTS = [
    {"service": "auth-service", "url": "https://example.com/auth/status"},
    {"service": "order-service", "url": "https://example.com/order/health"},
    {"service": "config-service", "url": "https://example.com/config/ping"},
]


def request_json(url, token, timeout=3):
    headers = {"Authorization": token}
    response = requests.get(url, headers=headers, timeout=timeout)
    response.raise_for_status()
    return response.json()


def check_auth_service(data):
    return data.get("code") == 0


def check_order_service(data):
    return data.get("status") == "UP"


def check_config_service(data):
    return data.get("success") is True


CHECKERS = {
    "auth-service": check_auth_service,
    "order-service": check_order_service,
    "config-service": check_config_service,
}


def build_result(service, ok, message):
    return {
        "service": service,
        "ok": ok,
        "message": message,
    }


def probe_one(endpoint, token):
    service = endpoint["service"]
    url = endpoint["url"]

    try:
        data = request_json(url, token)
    except requests.RequestException as exc:
        return build_result(service, False, str(exc))

    checker = CHECKERS[service]
    ok = checker(data)
    return build_result(service, ok, "ok" if ok else "check failed")


def run_probe(token):
    results = []
    for endpoint in ENDPOINTS:
        results.append(probe_one(endpoint, token))
    return results


def print_summary(results):
    failed = [item for item in results if not item["ok"]]
    print(f"巡检完成，总接口数: {len(results)}")
    print(f"成功: {len(results) - len(failed)}")
    print(f"失败: {len(failed)}")
    if failed:
        print("\n失败接口:")
        for item in failed:
            print(f'- {item["service"]}: {item["message"]}')


def save_report(results, output_file):
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(json.dumps(results, ensure_ascii=False, indent=2))

这里真正重要的不是“函数数量变多了”，而是职责开始稳定了：

  • request_json() 只管请求
  • check_xxx() 只管业务判断
  • build_result() 只管统一输出
  • probe_one() 负责把一次巡检串起来
  • print_summary() 负责展示

八、什么时候该从单文件继续往目录结构走

如果脚本只做一次性工作，单文件完全没问题。

但只要出现下面这些信号，就应该考虑继续拆：

  • 接口数量越来越多
  • 需要支持多个环境
  • 要加命令行参数
  • 要加测试
  • 要加不同的输出格式

这时更适合至少拆成这种目录：

1
2
3
4
5
6
7
8
api_probe/
├── main.py
├── config.py
├── client.py
├── checkers.py
├── models.py
├── report.py
└── tests/

分工大概可以这样定：

  • config.py：环境、token、超时配置
  • client.py：HTTP 请求封装
  • checkers.py：不同服务的判断规则
  • models.py：统一结果结构
  • report.py：输出和归档
  • main.py：主流程入口

一个实际错误：目录拆了，但职责没拆

有些项目虽然已经有多个文件，但内容仍然是：

  • client.py 里也在做业务判断
  • report.py 里也在拼请求参数
  • main.py 里还堆着各种细节

这种拆法只是把乱代码切成几段，不是真正的工程化。

九、把数据处理脚本也放进来一起看

接口工具和数据处理脚本看起来像两类东西，实际失控方式很像。

例如巡检跑完以后，要把结果清洗成一份只保留失败项的报告：

1
2
3
4
5
6
7
8
9
def collect_failed_services(results):
    failed = []
    for item in results:
        if not item["ok"]:
            failed.append({
                "service": item["service"],
                "reason": item["message"],
            })
    return failed

这类清洗逻辑最适合单独放函数，不要直接写在主流程里。

一个实际错误：主流程里边请求边清洗边输出

错误写法通常像这样：

1
2
3
4
5
6
7
8
9
for endpoint in ENDPOINTS:
    data = request_json(endpoint["url"], token)
    if endpoint["service"] == "auth-service":
        ...
    if ...:
        ...
    print(...)
    with open(...):
        ...

这类代码的最大问题不是长，而是改动成本极高。

一旦要加一个新字段，可能需要同时改：

  • 请求阶段
  • 判断阶段
  • 清洗阶段
  • 输出阶段

所以“数据处理脚本不要越写越乱”的核心，不是把代码写短，而是把数据流转阶段拆清。

十、完整版本放在一起

下面给出一个可以直接运行和继续扩展的完整版本。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
import json
import requests


ENDPOINTS = [
    {"service": "auth-service", "url": "https://example.com/auth/status"},
    {"service": "order-service", "url": "https://example.com/order/health"},
    {"service": "config-service", "url": "https://example.com/config/ping"},
]


def request_json(url, token, timeout=3):
    headers = {"Authorization": token}
    response = requests.get(url, headers=headers, timeout=timeout)
    response.raise_for_status()
    return response.json()


def check_auth_service(data):
    return data.get("code") == 0


def check_order_service(data):
    return data.get("status") == "UP"


def check_config_service(data):
    return data.get("success") is True


CHECKERS = {
    "auth-service": check_auth_service,
    "order-service": check_order_service,
    "config-service": check_config_service,
}


def build_result(service, ok, message):
    return {
        "service": service,
        "ok": ok,
        "message": message,
    }


def probe_one(endpoint, token):
    service = endpoint["service"]
    url = endpoint["url"]

    try:
        data = request_json(url, token)
    except requests.RequestException as exc:
        return build_result(service, False, str(exc))

    checker = CHECKERS[service]
    ok = checker(data)
    return build_result(service, ok, "ok" if ok else "check failed")


def run_probe(token):
    results = []
    for endpoint in ENDPOINTS:
        results.append(probe_one(endpoint, token))
    return results


def collect_failed_services(results):
    failed = []
    for item in results:
        if not item["ok"]:
            failed.append({
                "service": item["service"],
                "reason": item["message"],
            })
    return failed


def print_summary(results):
    failed = collect_failed_services(results)
    print(f"巡检完成，总接口数: {len(results)}")
    print(f"成功: {len(results) - len(failed)}")
    print(f"失败: {len(failed)}")

    if failed:
        print("\n失败接口:")
        for item in failed:
            print(f'- {item["service"]}: {item["reason"]}')


def save_report(results, output_file):
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(json.dumps(results, ensure_ascii=False, indent=2))

十一、怎么测试这一层是不是真的掌握了

这一层不能只看懂，要自己改一遍。

可以直接做这些动作：

  1. 增加一个新接口检查器
  2. 把 token 改成从配置文件读取
  3. 把失败结果单独导出成 failed.json
  4. 加一个 --env 参数切换环境
  5. 给 probe_one() 和 collect_failed_services() 补最小测试

如果这些改动都能独立完成，说明这篇最核心的工程化拆分已经开始真正掌握。

十二、一个实际排错场景

这类脚本里一个非常常见的实际问题是：巡检结果里明明显示失败，但 failed.json 里却是空的。

这时排查顺序通常很直接：

  1. 先看统一结果结构里失败字段是不是写对了
  2. 再看清洗函数筛选的是 ok 还是 success
  3. 再看是不是某个地方手写了另一套字段名

例如前面结果结构是：

1
{"service": "config-service", "ok": False, "message": "timeout"}

但清洗函数里却写：

1
2
if not item["success"]:
    ...

这时通常会报：

1
KeyError: 'success'

或者直接逻辑跑偏。

这类问题的根因通常不是清洗算法复杂，而是输出结构没有统一。

十三、一个实际练习

可以直接把这一篇变成一个完整练习。

练习目标：做一个“测试环境接口巡检工具”。

要求：

  1. 至少检查 3 个接口
  2. 把接口定义和执行逻辑分开
  3. 把校验规则单独拆出来
  4. 把结果统一成固定结构
  5. 导出一份总报告和一份失败报告
  6. 至少补一个实际错误场景，例如字段名不统一或某个接口判断规则写错

如果这个练习能独立做完，说明接口工具和数据处理脚本如何避免失控，这一层已经开始真正会用了。

十四、这篇文章学完以后，下一步应该补什么

如果这一篇已经能跟着做完，下一步最适合继续补的是：

  1. Python 写 HTTP 接口、命令行工具和定时任务时，目录和职责怎么拆
  2. pytest 的 fixture、mock 和测试数据怎么组织
  3. Python 项目从脚本到工程时，包结构和依赖管理怎么继续规范

因为到这一步，脚本已经开始进入“一个工具会不断长大”的阶段了。接下来最容易卡住的，就是入口越来越多以后，代码边界还能不能守住。

十五、结语

Python 做接口工具和数据处理脚本时，真正决定后面会不会乱的，不是语言本身，而是最开始有没有把职责拆开。

只要先把这几层分清：

  • 请求怎么发
  • 成功怎么判
  • 结果怎么统一
  • 数据怎么清洗
  • 报告怎么输出

脚本就会稳很多。哪怕功能继续加，代码也还是有继续演进的空间。