Python：写 HTTP 接口、命令行工具和定时任务时，目录和职责怎么拆

发表于 2018-06-27 分类于 Python 本文字数： 2.3k 阅读时长 ≈ 8 分钟

同一个 Python 项目里一旦同时出现 HTTP 接口、命令行工具和定时任务，最容易失控的就是入口和职责。这篇文章围绕一个统一的告警中心小项目，讲清楚目录和职责怎么拆。

Python 项目一开始通常只有一个入口：

一个脚本
一个命令
一段主流程

问题往往出现在项目继续长大以后。比如同一套业务逻辑，后来同时需要：

暴露一个 HTTP 接口
提供一个命令行工具
跑一个定时任务

这时最容易出现的代码状态是：

HTTP 路由里直接写业务逻辑
CLI 命令又复制一份同样的逻辑
定时任务里再复制第三份
配置、数据模型、输出格式到处都不一致

结果就是功能明明是同一套，代码却变成三份。

这一篇直接围绕一个实际小项目展开：做一个统一的告警中心工具。

这个项目要支持三种入口：

HTTP 接口：查询最近告警
命令行工具：手动触发一次告警汇总
定时任务：定时扫描并生成告警摘要

用这个场景把最关键的问题讲清楚：

入口和业务逻辑为什么必须分开
目录怎么拆，才不会三种入口各管一套
哪些代码适合放 service 层，哪些适合放 adapters/entry 层

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

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

设计一个同时支持 HTTP、CLI、定时任务的最小项目结构
把入口代码和核心业务逻辑分开
判断什么代码该放在路由层，什么该放在 service 层
识别几类典型坏味道，例如不同入口各写一套查询逻辑、响应格式不统一

如果这些动作能独立做出来，Python 项目出现多个入口时就不会那么容易失控。

二、先看这个项目要长成什么样

目标目录先放出来：

alert_center/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── services/
│   │   └── alert_service.py
│   ├── repositories/
│   │   └── alert_repo.py
│   ├── web/
│   │   └── routes.py
│   ├── cli/
│   │   └── commands.py
│   ├── jobs/
│   │   └── scan_job.py
│   └── schemas.py
└── tests/

这个目录的核心思想不是“文件越多越工程化”，而是：

不同入口放在不同位置
共同业务逻辑只保留一份

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

如果不拆职责，第一版通常长这样：

def get_recent_alerts():
    data = [
        {"service": "auth", "level": "warn"},
        {"service": "order", "level": "error"},
    ]
    return data


def http_handler():
    data = get_recent_alerts()
    return {"items": data}


def cli_command():
    data = get_recent_alerts()
    print(data)


def cron_job():
    data = get_recent_alerts()
    print("scan done", data)

这段代码短期看似乎没问题，但只要继续加功能，就会碰到：

HTTP 响应格式要变
CLI 想加过滤参数
定时任务想写数据库或文件

这时候三种入口很快就会开始各写各的。

四、第一层先拆：入口和业务逻辑分开

1. 什么叫入口层

入口层只负责：

接收输入
调用业务逻辑
把结果按当前入口需要的格式输出

它不应该直接承担：

核心查询逻辑
数据拼装逻辑
复杂校验逻辑

2. 先抽一层 service

app/services/alert_service.py：

def list_recent_alerts():
    return [
        {"service": "auth", "level": "warn"},
        {"service": "order", "level": "error"},
    ]

这样三种入口都可以共用：

HTTP 调它
CLI 调它
定时任务也调它

3. 一个实际错误：路由里直接写业务

错误写法通常像这样：

def http_handler():
    data = [
        {"service": "auth", "level": "warn"},
        {"service": "order", "level": "error"},
    ]
    return {"items": data}

这样一开始图快，后面 CLI 和 Job 一上来，就会复制一份类似逻辑。

五、第二层再拆：数据来源和业务处理也要分开

如果 alert_service.py 里以后还要接数据库、文件或者第三方接口，再继续往下拆一层。

1. 仓储层只负责取数据

app/repositories/alert_repo.py：

def fetch_alerts():
    return [
        {"service": "auth", "level": "warn"},
        {"service": "order", "level": "error"},
    ]

2. service 层负责业务处理

app/services/alert_service.py：

from app.repositories.alert_repo import fetch_alerts


def list_recent_alerts():
    alerts = fetch_alerts()
    return alerts


def count_error_alerts():
    alerts = fetch_alerts()
    total = 0
    for item in alerts:
        if item["level"] == "error":
            total += 1
    return total

这样以后数据来源换了，优先改 repo；业务规则变了，优先改 service。

六、三种入口分别该放什么

1. HTTP 接口层

HTTP 层最适合放：

请求参数解析
调用 service
返回 HTTP 格式结果

app/web/routes.py：

from app.services.alert_service import list_recent_alerts


def get_recent_alerts_api():
    data = list_recent_alerts()
    return {
        "items": data,
        "total": len(data),
    }

2. CLI 层

CLI 层最适合放：

解析命令行参数
调用 service
打印适合终端阅读的内容

app/cli/commands.py：

from app.services.alert_service import list_recent_alerts


def run_list_alerts():
    data = list_recent_alerts()
    print(f"总告警数: {len(data)}")
    for item in data:
        print(f'- {item["service"]}: {item["level"]}')

3. 定时任务层

Job 层最适合放：

调度入口
调用 service
写归档、打日志、发通知

app/jobs/scan_job.py：

from app.services.alert_service import count_error_alerts


def run_scan_job():
    total = count_error_alerts()
    print(f"error alerts: {total}")

4. 一个实际错误：三个入口都各自做统计

错误写法：

HTTP 自己算一次错误数量
CLI 再算一次
定时任务再算一次

这会直接导致：

逻辑重复
一改规则三处都得改

正确方向是：统计逻辑只放在 service 层一份。

七、再补一层：统一结构不要入口各自拼

如果每个入口都自己发明一套数据结构，后面又会开始乱。

例如：

HTTP 用 items
CLI 叫 alerts
Job 叫 records

更适合的做法是先在内部统一结构，再由入口层决定如何展示。

app/schemas.py：

def build_alert(service, level):
    return {
        "service": service,
        "level": level,
    }

app/repositories/alert_repo.py：

from app.schemas import build_alert


def fetch_alerts():
    return [
        build_alert("auth", "warn"),
        build_alert("order", "error"),
    ]

这样至少项目内部的数据结构是一致的。

八、真正能跑的完整版本

下面给出一个最小但完整的版本。

app/schemas.py：

def build_alert(service, level):
    return {
        "service": service,
        "level": level,
    }

app/repositories/alert_repo.py：

from app.schemas import build_alert


def fetch_alerts():
    return [
        build_alert("auth", "warn"),
        build_alert("order", "error"),
    ]

app/services/alert_service.py：

from app.repositories.alert_repo import fetch_alerts


def list_recent_alerts():
    return fetch_alerts()


def count_error_alerts():
    alerts = fetch_alerts()
    total = 0
    for item in alerts:
        if item["level"] == "error":
            total += 1
    return total

app/web/routes.py：

from app.services.alert_service import list_recent_alerts


def get_recent_alerts_api():
    data = list_recent_alerts()
    return {
        "items": data,
        "total": len(data),
    }

app/cli/commands.py：

from app.services.alert_service import list_recent_alerts


def run_list_alerts():
    data = list_recent_alerts()
    print(f"总告警数: {len(data)}")
    for item in data:
        print(f'- {item["service"]}: {item["level"]}')

app/jobs/scan_job.py：

from app.services.alert_service import count_error_alerts


def run_scan_job():
    total = count_error_alerts()
    print(f"error alerts: {total}")

app/main.py：

from app.cli.commands import run_list_alerts
from app.jobs.scan_job import run_scan_job
from app.web.routes import get_recent_alerts_api


def main():
    print("CLI 输出:")
    run_list_alerts()

    print("\nJob 输出:")
    run_scan_job()

    print("\nHTTP 输出:")
    print(get_recent_alerts_api())


if __name__ == "__main__":
    main()

执行：

1	python -m app.main

输出：

CLI 输出:
总告警数: 2
- auth: warn
- order: error

Job 输出:
error alerts: 1

HTTP 输出:
{'items': [{'service': 'auth', 'level': 'warn'}, {'service': 'order', 'level': 'error'}], 'total': 2}

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

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

可以直接做这些动作：

给 HTTP 接口增加一个按 level 过滤的参数
给 CLI 增加一个只看 error 的命令
让定时任务把统计结果写入文件
给 alert_service.py 补最小测试
再加一个入口，例如“导出 CSV”

如果这些改动都能独立完成，说明多入口项目的职责拆分已经开始真正掌握。

十、一个实际排错场景

这类项目里一个非常常见的实际问题是：HTTP 接口看到 2 条告警，CLI 却只看到 1 条。

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

先看三种入口是不是都调用了同一个 service
再看有没有某个入口自己额外做了过滤
再看仓储层返回的数据是不是统一

如果最后发现：

HTTP 调的是 list_recent_alerts()
CLI 却直接自己写了一套查询逻辑

那根因通常就不是数据源有问题，而是入口层绕过了统一 service。

十一、一个实际练习

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

练习目标：做一个“统一任务中心”小项目。

要求：

同时支持 HTTP、CLI、Job 三种入口
三种入口共用同一份 service 逻辑
仓储层和 service 层分开
输出结构统一
至少补一个入口层重复逻辑的错误场景
运行入口统一用模块方式执行

如果这个练习能独立做完，说明多入口项目的目录和职责拆分已经开始真正会了。

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

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

pytest 的 fixture、mock 和测试数据怎么组织
Python 性能优化里哪些地方最先值得看
项目继续长大以后，配置、日志和依赖注入怎么整理

因为到这一步，项目已经不只是一个脚本，而是开始同时服务多种运行方式了。接下来最容易卡住的，是测试和维护成本会不会失控。

十三、结语

Python 项目一旦同时出现 HTTP、CLI 和定时任务，真正关键的不是入口有几个，而是核心逻辑有没有只有一份。

只要先把这几层守住：

入口只做输入输出
service 只做业务
repo 只做取数
内部结构统一

项目就算继续长大，也不会那么容易变成三套代码并行演化。