Python 工程实践记录：从脚本到可维护服务

脚本阶段的问题会被放大#

Python 脚本最开始通常很简单。读取一个文件，处理一批数据，调用一次外部服务，写出结果。这样的代码写在一个文件里很自然，甚至没有必要提前设计什么架构。问题出现在脚本开始承担更多责任之后。

比如一个脚本原来只需要手动跑，后来变成定时任务；原来只处理一个输入文件，后来要处理不同来源；原来失败了看一眼终端就行，后来需要在后台排查；原来只给自己用，后来其他模块也要调用。这个时候，如果代码仍然把配置、逻辑、输入输出和日志混在一起，就会很难改。

1
def run_job(path: str) -> None:
2
    rows = load_rows(path)
3
    cleaned = [normalize(row) for row in rows]
4
    save_rows(cleaned)

这段代码看起来没问题，但它省略了很多工程问题：输入来源是否可靠，输出位置怎么配置，失败时怎么反馈，处理过程怎么记录，数据量变大怎么办。脚本阶段这些问题可以靠人盯着，服务阶段就不能。

我通常会在项目开始变大时做一次很小的拆分：入口只负责解析参数和启动流程，核心逻辑放进可测试函数，配置单独建模，外部依赖包在边界层。这样不会一下子变重，但能让后续维护容易很多。

1
from dataclasses import dataclass
2
from pathlib import Path
3

4

5
@dataclass(frozen=True)
6
class JobConfig:
7
    input_path: Path
8
    output_path: Path
9
    batch_size: int = 500
10

11

12
def run_job(config: JobConfig) -> int:
13
    count = 0
14
    for batch in read_batches(config.input_path, config.batch_size):
15
        rows = [normalize(row) for row in batch]
16
        write_batch(config.output_path, rows)
17
        count += len(rows)
18
    return count

这里没有引入复杂框架，只是把配置从散落变量变成了一个明确对象，把函数返回值变成了可观察结果。脚本仍然可以调用它，服务也可以调用它，测试也能直接跑它。

脚本到服务的变化还有一个很明显的信号：它开始需要“可重复”。手动运行时，很多步骤可以靠记忆，比如先准备目录，再执行命令，再把结果传到别处。服务化之后，这些步骤都要进入代码或部署流程。输入在哪里，输出到哪里，失败后能不能重跑，重复执行会不会产生脏数据，这些问题都要被认真处理。

我会尽量让任务逻辑具备幂等倾向。不是所有任务都能完全幂等，但至少要避免重复运行时造成不可理解的结果。比如写文件前先写临时结果，确认完成后再替换；处理批次时记录批次标识；外部调用失败时保存状态而不是直接丢掉上下文。Python 很适合快速串流程，但流程越长，越需要把这些中间状态设计清楚。

1
def write_result(target: Path, content: str) -> None:
2
    temporary = target.with_suffix(target.suffix + ".tmp")
3
    temporary.write_text(content, encoding="utf-8")
4
    temporary.replace(target)

这段代码只是一个很小的习惯：不要把未完成结果直接写到目标位置。服务运行时，进程可能中断，磁盘可能满，任务可能被重试。很多稳定性不是靠大框架提供，而是靠这些细节一点点堆出来。

项目结构要服务阅读和测试#

Python 项目很容易被目录结构拖住。太扁平，所有东西都堆在一起；太复杂，刚开始就像大系统。我的习惯是按边界拆，而不是按“工具类”“公共函数”这种模糊名词拆。

一个轻量服务可以大致分成几块：入口、配置、核心业务、外部适配、任务、测试。入口负责把请求或命令转成调用；核心业务不直接关心框架；外部适配负责文件、网络、数据库或模型调用；测试尽量围绕核心业务写。

1
app/
2
  main.py
3
  config.py
4
  domain/
5
    jobs.py
6
    models.py
7
  adapters/
8
    storage.py
9
    notifier.py
10
  tasks/
11
    worker.py
12
tests/
13
  test_jobs.py

这个结构不是标准答案，只是表达一种思路：能被业务复用的逻辑不要锁死在框架入口里。很多 Python 服务后来难测，是因为核心逻辑直接写在路由函数、命令回调或定时任务里。想测一段逻辑，必须启动整个环境，成本自然就高。

如果使用 FastAPI，我也会尽量让路由层保持薄一点。路由负责校验输入、调用服务、返回结果；真正处理逻辑的函数放到别处。

1
from fastapi import FastAPI, HTTPException
2
from pydantic import BaseModel
3

4
app = FastAPI()
5

6

7
class JobRequest(BaseModel):
8
    source: str
9
    dry_run: bool = False
10

11

12
class JobResponse(BaseModel):
13
    accepted: bool
14
    task_id: str
15

16

17
@app.post("/jobs")
18
def create_job(payload: JobRequest) -> JobResponse:
19
    try:
20
        task_id = submit_job(payload.source, dry_run=payload.dry_run)
21
    except ValueError as exc:
22
        raise HTTPException(status_code=400, detail=str(exc)) from exc
23

24
    return JobResponse(accepted=True, task_id=task_id)

这段代码里，路由没有直接处理任务细节。submit_job 可以被测试，也可以被命令行入口复用。框架是入口，不应该变成所有逻辑的容器。

命令行入口也值得保留。很多服务上线后，仍然会需要一些离线能力：重建索引、清理历史、导入数据、检查配置、补跑任务。如果这些能力只能通过 Web 入口触发，排查和维护会很不方便。更好的方式是核心逻辑保持独立，命令行和服务入口都调用同一层。

1
def main() -> None:
2
    config = load_config()
3
    count = run_job(config)
4
    print(f"processed {count} rows")
5

6

7
if __name__ == "__main__":
8
    main()

这个入口很普通，但它让服务之外还有一条维护路径。服务负责长期运行，命令行负责一次性维护动作，两者共享核心逻辑。这样既不会让服务路由变成工具箱，也不会让维护脚本复制一套业务代码。

类型标注不是装饰#

Python 是动态语言，但类型标注很有价值。它不是为了把 Python 变成 Java，也不是为了让每一行都写得很重。它的价值在于把函数边界说清楚，让工具能提前发现一部分错误，也让读代码的人少猜。

1
from typing import Iterable
2

3

4
def average(values: Iterable[float]) -> float:
5
    total = 0.0
6
    count = 0
7

8
    for value in values:
9
        total += value
10
        count += 1
11

12
    if count == 0:
13
        raise ValueError("values is empty")
14

15
    return total / count

这个函数的类型很简单，但读者知道它不要求列表，只要可迭代的数字即可。空输入会失败，失败方式也写出来了。类型标注和异常语义结合起来，接口就清楚很多。

数据结构可以用 dataclass，也可以用 Pydantic。dataclass 适合内部对象，轻量直接；Pydantic 适合外部输入，校验和转换能力更强。不要把所有东西都统一成一种模型。外部输入需要严格校验，内部流转更关注清晰和性能。

1
from pydantic import BaseModel, Field
2

3

4
class ImageTask(BaseModel):
5
    prompt: str = Field(min_length=1, max_length=2000)
6
    width: int = Field(ge=256, le=2048)
7
    height: int = Field(ge=256, le=2048)

这种模型很适合服务入口。它让输入约束有明确位置，错误也能自然反馈。相比在函数里手写一堆散落判断，可维护性更好。

类型检查工具也值得纳入流程。Python 不会强制你通过类型检查，但在工程里，能跑一下就跑一下。它不能发现所有问题，却能拦下一些拼写错误、空值误用和接口变更。

类型标注还会影响重构体验。Python 项目没有类型时，改一个字段名或函数返回值，很难知道影响面。全靠搜索不可靠，因为动态调用、字典字段和字符串拼接都会让搜索结果不完整。有了类型之后，工具至少能帮你指出一部分调用方。它不是绝对安全，但能把重构从纯手工变成半自动。

我不追求所有地方都写很复杂的泛型。很多时候，简单标注就足够。函数入参、返回值、核心数据结构、外部接口模型，这些地方写清楚，收益已经很明显。内部临时变量可以让工具推断，不需要为了形式把代码写得很满。

1
type UserId = str
2

3

4
def find_user_name(users: dict[UserId, str], user_id: UserId) -> str | None:
5
    return users.get(user_id)

这种别名没有改变运行时行为，但能让读者知道这个字符串不是普通文本，而是一个标识。Python 的类型系统是辅助工具，用得克制时很舒服，用得过度也会让代码变重。

配置要集中，环境差异要清楚#

脚本里最容易散落的是配置。路径写在函数里，超时时间写在调用处，开关写在全局变量里，等到要部署到不同环境时，就开始到处改。Python 项目如果想长期维护，配置要尽早集中。

配置集中不代表所有值都堆进一个巨大对象。更好的方式是按用途分组，同时让默认值清楚。敏感值和机器相关内容不要写进文章或代码示例里，真实项目中也要放在合适的运行环境里管理。

1
from dataclasses import dataclass
2

3

4
@dataclass(frozen=True)
5
class ServerConfig:
6
    host: str = "0.0.0.0"
7
    port: int = 8000
8
    workers: int = 2
9

10

11
@dataclass(frozen=True)
12
class AppConfig:
13
    server: ServerConfig
14
    task_timeout_seconds: int = 120
15
    log_level: str = "info"

配置对象的好处是可传递、可测试、可替换。函数不需要到处读全局变量，测试时也能构造一个小配置。配置读取可以单独做，业务逻辑只接收已经整理好的结果。

配置还要考虑失败。缺少必要配置时，应该在启动阶段报错，而不是运行到一半才失败。配置值不合法时，要有明确提示。服务启动失败并不可怕，悄悄带着错误配置跑起来才可怕。

配置也要区分“运行环境”和“业务参数”。运行环境决定服务怎么启动，比如监听地址、日志级别、工作进程；业务参数决定任务怎么处理，比如批次大小、超时时间、默认格式。两类配置混在一起，后面会很难管理。尤其是需要给任务做不同策略时，如果所有开关都散落在环境读取逻辑里，测试和复现都会变麻烦。

我更愿意在启动阶段把原始配置读出来，转换成明确对象，然后把对象传给需要的模块。这样业务逻辑不关心配置来自哪里。测试时可以直接构造配置，不必模拟完整运行环境。

1
def build_config(raw: dict[str, str]) -> AppConfig:
2
    return AppConfig(
3
        server=ServerConfig(
4
            port=int(raw.get("PORT", "8000")),
5
            workers=int(raw.get("WORKERS", "2")),
6
        ),
7
        task_timeout_seconds=int(raw.get("TASK_TIMEOUT", "120")),
8
    )

示例里只展示结构，不代表真实项目要把所有值都这样处理。重点是配置转换有固定位置，错误能在启动时暴露，业务函数拿到的是已经整理过的配置。

日志不是 print 的替代品#

Python 写脚本时用 print 很自然，但服务需要更稳定的日志。日志要能区分级别，要能带上下文，要能被收集。服务出了问题时，日志通常是排查入口。

1
import logging
2
from time import perf_counter
3

4
logger = logging.getLogger(__name__)
5

6

7
def process_batch(batch_id: str, rows: list[dict[str, object]]) -> int:
8
    started = perf_counter()
9
    logger.info("batch started", extra={"batch_id": batch_id, "size": len(rows)})
10

11
    count = handle_rows(rows)
12

13
    elapsed_ms = int((perf_counter() - started) * 1000)
14
    logger.info("batch finished", extra={"batch_id": batch_id, "elapsed_ms": elapsed_ms})
15
    return count

这里的重点不是使用哪个日志库，而是日志里要有上下文。只写“start”和“done”帮助不大。任务标识、输入规模、耗时、失败原因，这些信息能让排查快很多。

异常日志也要克制。不要捕获所有异常后只输出一句话，也不要把异常吞掉继续跑。能恢复的错误就明确恢复，不能恢复的错误就让上层知道。宽泛捕获应该非常小心。

1
def safe_load(path: Path) -> dict[str, object]:
2
    try:
3
        return load_json(path)
4
    except OSError as exc:
5
        logger.warning("file read failed", extra={"path": str(path), "reason": str(exc)})
6
        raise

这段代码没有假装处理掉错误，只是补充了上下文，然后继续抛出。很多时候，这比“捕获后返回空字典”安全得多。返回空结果可能会让后面的逻辑误以为数据真实为空，问题被藏得更深。

日志还要考虑关联。一次请求、一个任务、一个批次，最好有稳定标识贯穿整个流程。没有这个标识时，日志虽然很多，但很难串起来。尤其是异步任务和后台 worker，开始日志、处理中日志、失败日志如果没有同一个任务标识，排查时就像在拼碎片。

1
def run_recorded_job(task_id: str) -> None:
2
    logger.info("task started", extra={"task_id": task_id})
3
    try:
4
        run_task_body(task_id)
5
    except Exception:
6
        logger.exception("task failed", extra={"task_id": task_id})
7
        mark_task_failed(task_id)
8
        raise
9
    else:
10
        mark_task_succeeded(task_id)
11
        logger.info("task succeeded", extra={"task_id": task_id})

这里捕获了宽泛异常，但边界很明确：后台任务的最外层。它记录异常，更新任务状态，然后继续抛出，让运行器也能感知失败。宽泛捕获如果放在内部业务函数里，容易吞掉问题；放在任务边界，则可以作为统一记录点。

测试要从核心逻辑开始#

Python 的测试体验很好，但很多项目仍然没有测试。原因往往不是工具不好，而是代码结构不好测。逻辑写在入口里，依赖直接访问外部资源，配置从全局读取，测试自然麻烦。

测试应该从核心逻辑开始。纯函数、数据转换、校验规则、错误路径，这些都很适合先测。外部依赖可以通过适配层隔开，测试时换成假的实现。

1
def normalize_name(value: str) -> str:
2
    return " ".join(value.strip().lower().split())
3

4

5
def test_normalize_name() -> None:
6
    assert normalize_name("  Firefly   Blog ") == "firefly blog"

这种测试很小，但它能保护基础行为。项目变大后，很多错误都来自这些不起眼的转换逻辑。把它们测住，重构时会安心很多。

服务入口也可以测，但不要让所有测试都变成端到端测试。端到端测试有价值，但慢，定位也不方便。更好的组合是：核心逻辑大量小测试，关键路由少量集成测试，耗时任务做独立测试。

1
def test_run_job_writes_clean_rows(tmp_path: Path) -> None:
2
    source = tmp_path / "input.txt"
3
    target = tmp_path / "output.txt"
4
    source.write_text(" Alice \nBOB\n", encoding="utf-8")
5

6
    config = JobConfig(input_path=source, output_path=target, batch_size=2)
7
    count = run_job(config)
8

9
    assert count == 2
10
    assert target.read_text(encoding="utf-8").splitlines() == ["alice", "bob"]

tmp_path 让文件测试很自然，不需要依赖真实目录。测试越接近真实行为，越能发现问题；测试越少依赖真实环境，越容易稳定运行。

测试外部依赖时，我会优先使用假的适配器，而不是在每个测试里 mock 很深的内部调用。mock 太深会让测试和实现绑死，重构时很容易碎。适配器边界如果清楚，测试可以换成内存实现。

1
class MemoryStorage:
2
    def __init__(self) -> None:
3
        self.items: dict[str, str] = {}
4

5
    def write(self, key: str, value: str) -> None:
6
        self.items[key] = value
7

8
    def read(self, key: str) -> str | None:
9
        return self.items.get(key)

这种假的实现很简单，但它验证的是业务行为，不是具体外部服务。真实适配器可以单独做少量集成测试。这样测试层次会更清楚：核心逻辑快而多，外部集成少而关键。

异步和任务要看场景#

Python 服务很容易遇到耗时任务。图片处理、文件转换、外部服务调用、批量数据处理，都不适合直接卡在请求里。这个时候要把请求响应和任务执行分开。

异步不是万能答案。async 很适合 I/O 密集场景，比如大量网络等待；如果是 CPU 密集任务，单纯改成 async 不会让它更快。需要根据任务性质选择线程池、进程池、队列或独立 worker。

1
from concurrent.futures import ThreadPoolExecutor
2

3
executor = ThreadPoolExecutor(max_workers=4)
4

5

6
def submit_job(source: str, dry_run: bool = False) -> str:
7
    task_id = create_task_record(source=source, dry_run=dry_run)
8
    executor.submit(run_recorded_job, task_id)
9
    return task_id

这个例子很轻量，适合小工具或低并发服务。更复杂的场景可以换成更完整的队列系统，但思路一样：请求提交任务，任务后台执行，前端或调用方查看状态。

任务化之后，状态就很重要。任务是等待中、处理中、完成、失败，应该有明确记录。失败原因也要保存，不然用户只看到“失败”两个字，维护者也不知道问题发生在哪。

1
class TaskState(str, Enum):
2
    pending = "pending"
3
    running = "running"
4
    succeeded = "succeeded"
5
    failed = "failed"

状态不要设计得过度复杂，但要足够表达流程。后续如果需要重试、取消或清理，再逐步补。任务系统一开始最重要的是可观察：知道它在做什么，知道它为什么失败。

性能边界要诚实#

Python 性能不差，但它有边界。很多服务慢，不是因为 Python 本身，而是因为 I/O 等待、数据库访问、网络调用、文件处理、序列化成本。也有一些场景确实会被 CPU 算力限制。关键是把问题分清楚。

性能排查不要靠感觉。可以先用简单计时，把耗时拆开：读取花多久，处理花多久，写出花多久，外部调用花多久。等知道瓶颈在哪，再决定要不要优化。

1
from contextlib import contextmanager
2
from time import perf_counter
3

4

5
@contextmanager
6
def measure(name: str):
7
    started = perf_counter()
8
    try:
9
        yield
10
    finally:
11
        elapsed_ms = int((perf_counter() - started) * 1000)
12
        logger.info("step measured", extra={"step": name, "elapsed_ms": elapsed_ms})

这种小工具很朴素，但能让排查有数据。很多时候，你以为慢在处理逻辑，实际慢在网络；你以为慢在读取文件，实际慢在写日志或重复解析配置。

数据处理时，Python 也要注意内存。能流式处理就不要一次把所有数据读进内存。能分批就分批。列表推导很好用，但面对大数据时要知道它会创建完整列表。

1
def iter_clean_rows(path: Path) -> Iterator[str]:
2
    with path.open("r", encoding="utf-8") as file:
3
        for line in file:
4
            value = normalize_name(line)
5
            if value:
6
                yield value

生成器让数据按需流动。它不一定适合所有场景，但处理大文件时很有帮助。Python 的工程化不只是框架，也包括这些基础习惯。

如果遇到真正的 CPU 密集任务，可以考虑把热点交给更合适的库，或者拆成独立服务。Python 很适合编排，不一定要自己承担所有计算。能用成熟库就不要手写低效循环；能把热点边界清楚地拆出来，后续也更容易替换。

性能边界还包括序列化和对象创建。很多服务的耗时不是算法本身，而是大量字典和模型来回转换。外部入口需要校验模型，内部处理不一定每一步都要重新包一层对象。数据结构要根据边界选择：入口清晰，内部轻量，输出稳定。Python 的灵活性很好，但灵活不代表每一层都用无结构字典。

1
def to_response(task: TaskRecord) -> dict[str, object]:
2
    return {
3
        "id": task.id,
4
        "state": task.state,
5
        "created_at": task.created_at.isoformat(),
6
    }

输出转换集中写，后面调整响应格式会容易很多。不要让路由函数、任务函数和存储函数各自拼一份结果。重复不只影响代码量，也会让字段语义慢慢分叉。

部署后才是真维护#

Python 服务能在本机启动，不代表部署完成。部署之后还有依赖版本、启动命令、日志位置、进程管理、健康状态、数据目录、备份和升级。很多问题不是代码逻辑，而是运行方式不稳定。

依赖要固定版本。今天能装，不代表半年后还能装出同样结果。无论使用哪种工具，都应该有锁定文件或明确版本策略。依赖升级也不要混在功能改动里，最好单独做，方便回滚和排查。

启动方式要写清楚。服务入口是什么，工作进程多少，静态资源怎么处理，任务 worker 怎么启动，配置从哪里来，这些都应该能被复现。不要只靠终端历史。

日志和错误也要能在服务器上看到。服务挂了，任务失败了，耗时变长了，如果没有日志和基本指标，只能猜。Python 项目的可维护性，很大一部分来自这些运行期信息。

发布流程也要有节奏。功能改动、依赖升级、运行环境调整，最好不要全部混在一起。Python 依赖生态更新很快，一个小版本变化就可能影响类型、校验或底层二进制包。把依赖升级单独做，跑完测试再发布，会比顺手升级稳很多。

数据备份和任务恢复也不能忽略。只要服务会写文件、生成结果或保存任务状态，就要考虑故障后的恢复方式。哪些数据可以重新生成，哪些数据必须备份，哪些任务可以重跑，哪些任务需要人工确认，这些判断要提前写进维护习惯里。

我也会给维护命令留位置。比如检查配置、列出失败任务、清理临时文件、重建缓存、导出诊断信息。这些命令平时不显眼，真正出问题时很有用。服务的可维护性，不只是主流程跑得好，也包括异常时有工具能处理。

Python 项目的维护还要面对一个很现实的问题：环境漂移。系统 Python 版本、虚拟环境、依赖缓存、本地工具、服务器镜像，任何一处变化都可能让“之前能跑”的服务出现差异。为了减少这种差异，我会尽量把运行命令、依赖安装、测试命令和构建过程写成固定脚本或容器流程。不是为了显得正规，而是为了下次迁移或恢复时不靠回忆。

1
def check_runtime(config: AppConfig) -> list[str]:
2
    problems: list[str] = []
3

4
    if config.server.workers < 1:
5
        problems.append("worker count must be positive")
6

7
    if config.task_timeout_seconds < 10:
8
        problems.append("task timeout is too small")
9

10
    return problems

这种运行前检查很普通，却能挡掉很多低级配置问题。服务启动时发现配置不对，比任务跑到一半才失败更好。Python 给了我们很高的动态能力，但动态能力也意味着很多错误要靠工程习惯提前发现。配置检查、依赖锁定、启动自检、基础测试，这些都是把动态系统变得更可控的方式。

还有一个值得持续维护的方向是数据格式兼容。脚本阶段，输出格式随手改一下也没关系；服务阶段，输出可能被页面、任务、其他脚本或人工流程依赖。字段改名、时间格式变化、状态值变化，都可能影响下游。哪怕只是自己的工具，也应该尽量让格式变化有节奏。需要调整时，保留一段兼容期，或者写清楚迁移方式，会比突然改掉稳很多。

我也会把“人工能不能接手”当成维护标准。很多 Python 工具刚写出来时只适合作者本人使用，参数名、文件名、日志内容都带着很强的临时感。等它变成服务后，维护者可能已经不是写这段代码的人，至少也可能是几个月后的自己。这个时候，命令输出是否清楚、错误提示是否能行动、任务状态是否能解释，就会变得很重要。好的 Python 工程不一定架构很大，但应该让人知道下一步该做什么。

还有一个很实际的判断：能不能在不打开源码的情况下完成常规维护。比如查看当前配置摘要、确认依赖版本、检查任务状态、重跑某个失败批次、清理过期产物，这些动作如果都必须靠改代码完成，服务就还停留在脚本阶段。Python 很适合快速补这些维护入口，把它们补齐之后，项目会更像一个可以长期放着运行的工具。

继续把脚本养成工程#

Python 的优势是轻快，但轻快不等于随意。脚本可以先跑起来，这是它的长处；当脚本开始承担稳定职责时，就要逐步补上工程边界。配置集中一点，类型明确一点，日志认真一点，测试覆盖一点，任务状态可见一点，部署流程固定一点，项目就会稳很多。

我不喜欢把 Python 项目一开始就做得很重。很多工具本来只是解决一个小问题，没必要套上复杂架构。但我也不喜欢一直停在临时脚本状态。比较舒服的方式是随着使用强度慢慢整理：重复逻辑抽出来，外部依赖包起来，错误路径补清楚，测试跟着关键行为走。

这篇记录后面也会继续维护。Python 生态变化很快，类型工具、打包方式、异步框架、测试工具、部署习惯都在变化。真正值得保留的不是某个具体库，而是这些判断：逻辑和入口分开，配置和代码分开，任务和请求分开，错误和成功都要可见。只要这些边界还在，项目就不容易乱。