FastAPI服务化重构：从失控脚本到可控服务的完整实战

先说说我为什么要折腾这件事。

之前有一个跑了大半年的 Python 脚本，每天凌晨从几个数据源拉取内容、清洗、算指标、把结果写进数据库。功能一直没出过大事故，但说实话，它离"让人放心"差得很远。我判断脚本是否正常的唯一方式，是去看最终产出的数据文件是不是新鲜的。有一天数据文件是空的，我甚至分不清是昨天凌晨就挂了还是今天刚坏——脚本里全是散落的print，Windows 计划任务里的输出早找不到了。那次之后我下定决心，要把这个脚本重构成一个真正的服务。

最终落地的是一个基于 FastAPI 的常驻进程，按照 API 层、Service 层、Repository 层的分层架构来组织。这篇文章把整个思考过程和实操细节完整写下来：为什么选 FastAPI 而不是 Flask 或 Django、目录结构怎么定、旧脚本的逻辑怎么拆、部署时踩了哪些坑——尤其是 uvicorn 日志丢失那个经典问题。如果你手上也有一堆"能跑但不可控"的脚本，正打算把它们服务化，这篇文章应该能给你一些直接的参考。

1. 脚本能跑就行？我先说说为什么非要改成服务

1.1 脚本时代的三个致命短板

很多人觉得脚本只要能出结果就行，没必要搞成什么服务。这个观点在"自己用、低频、出错了看一眼就明白"的阶段是成立的。但我的实际情况是：脚本已经被多个业务方间接依赖了，数据文件一空，下游报表、接口、其他人的分析全都会受影响，而我只能在出事之后被动发现。

我把脚本时代的痛点归纳成三个：

• 无观测：脚本跑完就退出，中间过程只剩print。在计划任务/后台任务里，这些输出要么丢失、要么被吞掉，出了问题只能"盲猜"。
• 无并发：脚本是单进程串行的。一旦有两个调用方同时触发，要么加锁、要么互相踩数据，怎么处理都别扭。
• 无边界：所有逻辑耦合在一个main()和几个"辅助函数"里。表面上看改动方便，实际上任何一处变化都会引发连锁反应，改完还得担心会不会影响其他环节。

用一个生活类比来说：脚本就像你家里的一个抽屉，钥匙、零钱、水电费单全放一起。平时拿东西确实方便，但有一天你想把"水电费单"单独交给另一个人去处理，就会发现所有东西纠缠在一起，根本没法单独分出去。脚本里的函数也一样，数据获取、清洗、计算、落库全挤在一个函数里，看着没毛病，可一旦有了"复用""交接""扩展"这些需求，就开始痛苦了。

1.2 出现这几种信号，说明改造时机到了

不是所有脚本都需要改造成服务。我给自己总结了几个"必须改造"的信号，你可以对照一下：

• 脚本开始被"别人"调用——这里的"别人"可能是另一个程序、另一台机器、或者一个网页按钮。
• 同一套逻辑在多个脚本里重复出现——获取数据、清洗、落库这些代码拷贝来拷贝去。
• 需要对外提供查询能力——比如同事问你"今天的数据跑出来了吗？结果是多少？"，你只能去翻数据库，没法给他一个入口。
• 答不上来"现在状态怎么样"的问题——脚本当前在跑哪个步骤、上次成功是什么时候、这次失败卡在哪一行，你完全没有概念。

一旦出现两条以上，就别再往脚本里打补丁了。我当时同时中了四条，所以彻底下决心重写。

1.3 为什么选 FastAPI，而不是 Flask 或 Django

选型这件事我认真比过，不是跟风。当时候选有 Flask、Django、FastAPI 三个，各自特点差异很大。

最终让我定下心来的是 FastAPI 的依赖注入和Pydantic 模型。依赖注入天然引导你拆分层——每个接口声明自己需要什么 Service，而不是在业务函数里到处 new 依赖。这跟"把脚本拆干净"的目标高度一致。哪怕你的脚本逻辑比我的还复杂，这个思路也适用。

2. 重构前夜：把旧脚本拆成一幅职责地图

2.1 动刀之前，先给旧脚本画图

这一步很容易被跳过，但恰恰是最关键的。我当时没有直接开写新代码，而是先把旧脚本的每一个函数、每一个流程列出来，按"获取数据 / 处理数据 / 输出数据"三个维度分组。

比如我这个脚本，本质上是三段式：

1. 从外部数据源拉取原始内容（HTTP 请求 + 解析）
2. 清洗、去重、计算指标（纯逻辑处理）
3. 把结果写进数据库（落库）

听起来很简单对不对？但当你真的去翻代码时，会发现这三件事互相穿插：拉数据时顺带做了格式整理，计算时又回头查了上一次的缓存，落库时还顺手改了一下数据内容。这种"穿插"就是脚本时代最大的技术债。

我的做法是：拿张纸或者用一个文档，把每个函数的输入、输出、副作用（比如修改了哪个全局变量、写了哪个文件）都标出来。这个动作花不了太长时间，但后面写分层代码时，你会省掉大量返工的痛苦。

2.2 识别隐藏依赖：全局变量、硬编码路径、隐式流程

脚本里的隐藏依赖是重构时最大的坑，比接口设计难得多。我当时总结了三类：

第一类是全局变量。脚本里常见的写法是模块顶部定义一堆DATA_DIR = "./data"、API_URL = "https://xxx"，然后在函数里直接用。这种"隐式共享状态"在脚本里很好用，但一旦拆成服务，多个请求并发访问时，全局变量就会变成竞态条件。我的处理原则是：所有配置项全部收敛到 Settings 类，所有需要共享的状态显式传入或存到数据库。

第二类是硬编码路径。原来脚本里到处是open("./data/result.csv")这种写法。改造后这绝对不能留，必须统一配置成环境变量或配置文件，否则服务一换目录就全崩。

第三类是隐式流程。脚本的main()里有一堆有顺序的操作，比如先检查数据源是否可用、再判断有没有昨天没跑完的任务、最后才拉数据。这个顺序是业务规则，在旧脚本里它只是"代码行数"，但在新架构里，它必须被显式地放到 Service 层的编排逻辑里，让读代码的人一眼就能看到业务是怎么流转的。

2.3 确定第一版接口范围，别贪多

我见过不少人重构时恨不得把所有功能都暴露成接口。这是大忌。第一版接口越少越好，先跑通"核心链路"比什么都重要。

我当时只定了三个接口：

• GET /api/v1/tasks/status：查任务当前状态（空闲、运行中、上次失败原因）
• POST /api/v1/tasks/run：手动触发一次执行
• POST /api/v1/tasks/config：修改执行参数（如数据源的 URL、阈值等）

这个范围很小，但它覆盖了脚本时代最痛的三件事：不可观测、不可手动触发、不可动态配置。至于那些复杂的数据查询接口，重构稳定之后再慢慢加，完全不迟。

3. 分层架构落地：一个能直接抄的 FastAPI 目录结构

3.1 最终采用的目录结构

网上关于 FastAPI 项目目录结构的讨论非常多，我根据自己的场景选了一套"不过度设计、但边界清晰"的结构。你可以直接参考：

app/ ├── main.py # 应用入口，创建 FastAPI 实例，注册路由 ├── core/ │ ├── config.py # Settings，基于 pydantic-settings │ ├── logging.py # 统一日志配置 │ └── exceptions.py # 自定义异常类 ├── api/ │ ├── __init__.py │ └── v1/ │ ├── router.py # v1 路由聚合 │ └── endpoints/ │ └── tasks.py # 任务相关接口 ├── services/ │ ├── task_service.py # 业务编排层，核心逻辑 │ └── data_process.py # 数据处理逻辑 ├── repositories/ │ └── task_repository.py # 数据访问层，只负责读写存储 ├── models/ │ └── task.py # 数据库 ORM 模型 ├── schemas/ │ └── task.py # Pydantic 模型，API 请求/响应结构 └── utils/ └── http_client.py # 通用 HTTP 请求封装

很多模板还会加routers/、dependencies/、middlewares/这些目录。如果你的服务接口数量多、依赖复杂，可以加；如果像我这个规模，上面前面那套已经绰绰有余。

3.2 各层职责与依赖方向

这套分层的核心，就是每一层只干一件事，并且依赖方向只能从上往下。

• API 层（api/v1/endpoints/）：接收 HTTP 请求，做参数校验（靠 Pydantic schema 自动完成），然后调用 Service 层的方法，把结果组装成响应返回。
• Service 层（services/）：放业务规则和流程编排。所有"什么时候该做什么事"都在这层体现。这一层不关心 HTTP、不关心数据库，只关心业务逻辑。
• Repository 层（repositories/）：放数据读写细节。查询数据库、写文件、调外部 API 这类操作统一封装在这一层。这样将来换数据库、换文件存储方式时，只需要改这一层。

依赖方向是API -> Service -> Repository，禁止反向。Service 不 import API 层的东西，Repository 不 import Service 层的东西。这个规则不费成本，但能保证你的代码不会在半年后又滚成一团。

3.3 用依赖注入把各层串起来

分层架构的代码结构只是骨架，真正让各层"松耦合"工作的是依赖注入。FastAPI 的Depends机制用起来非常方便，我在 Service 和 Repository 之间就是这么接的：

# repositories/task_repository.py class TaskRepository: def get_status(self) -> dict: # 查询任务状态的实际逻辑 return {"status": "idle"} # services/task_service.py class TaskService: def __init__(self, repo: TaskRepository): self._repo = repo def query_status(self) -> dict: # 业务逻辑可能在这里做缓存、做判断 return self._repo.get_status() # api/v1/endpoints/tasks.py from fastapi import APIRouter, Depends router = APIRouter() def get_task_service() -> TaskService: return TaskService(TaskRepository()) @router.get("/tasks/status") def get_task_status(service: TaskService = Depends(get_task_service)): return service.query_status()

这个写法看起来简单，但它有几个实打实的好处：

• 单元测试时，我可以随便构造一个假的TaskRepository传给TaskService，不需要真的数据库。
• 以后想加缓存、加监控，只需在get_task_service里改一行，接口完全不用动。
• 因为依赖关系是显式声明的，读代码的人一眼就能看出某个接口依赖哪些服务。

提示：在业务规模不大时，Depends不需要整得太花哨，直接按"构造 Service 并传入 Repository"这个最简单的模式来就行。复杂化交给以后真正需要的时候。

4. 动手迁移：从 main() 到 API 接口，我踩过的细节坑

4.1 同步逻辑改异步：哪些真需要 async

这是一个很常见的误区：用了 FastAPI 就恨不得把所有函数都写成async def。实际上，如果你的逻辑里有大量同步 I/O（比如requests.get、文件读写），把它们硬改成 async 反而会带来麻烦。

FastAPI 对同步函数有内置支持：用def定义的路由函数会被自动丢到线程池里执行，不会阻塞事件循环。也就是说，你完全可以先把原来的脚本函数直接搬到def路由里跑，性能没问题，代码改动也最小。

我在这次重构里的实际做法是分两步走。第一步，所有路由先用同步def，把功能跑通；第二步，再针对真正的高频 I/O 场景（比如查询状态接口，因为会被频繁探测）改成async def并配套使用httpx.AsyncClient。这样的渐进式迁移比一次性全改成异步要稳得多，也更容易排查问题。

4.2 配置管理：从"脚本顶部的常量"到 pydantic-settings

旧脚本的配置是一堆写在文件顶部的常量，比如：

DATA_DIR = "./data" API_URL = "https://example.com/data" TASK_INTERVAL = 3600

这在新架构里是绝对不能出现的。我全部迁移到了pydantic-settings管理的 Settings 类中：

# core/config.py from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): app_name: str = "data-service" data_dir: str = "./data" api_url: str = "https://example.com/data" task_interval: int = 3600 log_level: str = "INFO" model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", ) settings = Settings()

好处很明显：配置不再散落在各个文件里，而是统一从环境变量或.env文件读取。部署到新环境时，只需要换一份环境变量，不需要改代码。SettingsConfigDict里的env_file=".env"让本地开发也简单，改配置不用重新部署。

4.3 异常处理与统一响应

脚本时代处理异常的方式是"捕获之后 print 一下继续跑"，运气不好就静默失败。重构后我做了两件事：

第一，定义自己的异常基类，比如BusinessError，然后在全局异常处理器里把它转成统一的 HTTP 响应：

# core/exceptions.py class BusinessError(Exception): def __init__(self, code: str, message: str): self.code = code self.message = message # main.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from core.exceptions import BusinessError app = FastAPI() @app.exception_handler(BusinessError) async def business_error_handler(request: Request, exc: BusinessError): return JSONResponse( status_code=400, content={"code": exc.code, "message": exc.message}, )

第二，对外响应结构统一为{"code": 0, "data": ..., "message": "ok"}这种格式。这样做的好处是前端和后端协作时不用为每个接口猜返回结构。脚本改服务最怕无规则可循，统一响应格式成本低、收益大。

4.4 日志：脚本里的 print 在这套架构里怎么安置

旧脚本里到处都是print，我统计了一下可能有二十多个。改造后这些print全部删除，换成结构化日志：

# core/logging.py import logging import logging.config LOGGING_CONFIG = { "version": 1, "disable_existing_loggers": False, "formatters": { "default": { "format": "[%(asctime)s] %(levelname)s [%(name)s:%(lineno)s] %(message)s", }, }, "handlers": { "console": { "class": "logging.StreamHandler", "formatter": "default", }, "file": { "class": "logging.handlers.TimedRotatingFileHandler", "filename": "logs/data-service.log", "when": "midnight", "backupCount": 14, "formatter": "default", }, }, "loggers": { "uvicorn": {"handlers": ["console", "file"], "level": "INFO", "propagate": False}, "uvicorn.error": {"handlers": ["console", "file"], "level": "INFO", "propagate": False}, "uvicorn.access": {"handlers": ["console", "file"], "level": "INFO", "propagate": False}, "app": {"handlers": ["console", "file"], "level": "INFO", "propagate": False}, }, } logging.config.dictConfig(LOGGING_CONFIG)

业务代码里只需要：

logger = logging.getLogger("app") logger.info("开始拉取数据源 %s", api_url)

用日志对象代替print，最大的价值是可以按级别过滤、按时间滚动、输出到文件。出问题之后终于能说出"它是在哪一步挂的了"。

5. 让服务稳定跑起来：uvicorn、systemd 与日志那点事

5.1 uvicorn 怎么启动才符合生产要求

本地开发可以直接uvicorn app.main:app --reload，但部署到服务器上不能这么跑。我整理了我自己最终使用的启动命令：

uvicorn app.main:app \ --host 0.0.0.0 \ --port 8000 \ --workers 2 \ --limit-concurrency 256 \ --timeout-keep-alive 5

几个参数说明一下：

• --workers 2：开两个进程。因为我的服务没有太多共享内存状态，多进程能带来真实的并发提升。但要注意，如果你用了进程内缓存、状态变量，--workers大于 1 时要谨慎，多个进程之间不会共享这些状态。
• --limit-concurrency 256：限制最大并发请求数，防止被打爆。
• --timeout-keep-alive 5：保持连接超时设短一点，避免大量闲置连接占满文件描述符。

如果你是单核小机器，--workers 1也完全没问题。FastAPI 同步路由本身就在线程池里，单进程也能撑住中等并发量。

5.2 systemd 托管与开机自启

服务化之后，绝对不能依赖"手动登录服务器执行命令行"。我用 systemd 把它托管起来，这样能获得开机自启、崩溃自动重启、启动日志统一管理这些能力。

写一个 service 文件，放到/etc/systemd/system/data-service.service：

[Unit] Description=Data Service (FastAPI) After=network.target [Service] User=www-data WorkingDirectory=/opt/data-service EnvironmentFile=/etc/data-service.env ExecStart=/opt/data-service/venv/bin/uvicorn app.main:app --host 0.0.0.0 --port 8000 Restart=always RestartSec=3 [Install] WantedBy=multi-user.target

几个细节：

• EnvironmentFile用来加载环境变量文件，这样配置不必写死在 systemd 配置里。
• Restart=always是必须的，脚本时代最大的痛就是"挂了不知道"，现在服务挂了会自动拉起。
• WorkingDirectory必须正确，否则日志路径、配置相对路径全都对不上。

激活命令：

sudo systemctl daemon-reload sudo systemctl enable>@app.get("/health") def health_check(): # 这里可以检查数据库连接是否正常、外部依赖是否可用 return {"status": "alive"}

这个接口有两个用途：一是部署平台的探活；二是你自己快速确认服务状态，不用去翻日志。真正生产环境我还会往里面塞一点依赖状态检查，比如数据库能不能连、外部数据源通不通。这样负载均衡器或者监控系统就能在服务"带病运行"前及时知道。

另外，systemd 的Restart=always解决了崩溃重启的问题，但"优雅退出"同样重要：当服务收到终止信号时，应该让正在处理的请求跑完而不是立刻被掐断。uvicorn 本身对 SIGTERM 的处理已经比较完善，只要你不强制kill -9，它会给正在处理的请求一个宽限期。真正需要注意的反而是你的业务代码：别在进程退出逻辑里做太重的清理操作，否则会拖慢重启速度。

6. 重构之后的收益：以及我会怎么继续演进

6.1 三个立竿见影的改变

重构完成并稳定运行之后，有几个改变是立竿见影的。

第一是可观测性。现在任何时间点，我都能通过GET /api/v1/tasks/status知道任务在不在跑、上次结果怎么样。配合日志文件，出问题时定位速度从"小时级"变成了"分钟级"。

第二是可控性。之前脚本只在计划任务里定时跑，想手动触发一次就要登录服务器敲命令。现在直接发一个请求就行，甚至可以接一个简单的管理页面。数据源的地址变了，也只需要改配置或调用配置接口，不用改代码重新部署。

第三是可测试性。分层之后，每个 Service 方法都可以独立测。我用 pytest 写了针对核心业务逻辑的测试用例，跑一遍只要几秒钟。脚本时代根本没有这种体验，因为所有逻辑都耦合在一起，想测一个函数就得先让前面的步骤全部跑一遍。

6.2 从单体服务到微服务的演进空间

这次分层架构还有一个隐藏收益：它为将来拆微服务留好了路。比如后来我如果要把"数据拉取"和"指标计算"拆成两个独立服务，Service 层和 Repository 层的边界已经天然把它俩分开了，从接口调用变成 RPC/消息队列调用，改动范围是可控的。如果你一开始就是平铺的脚本代码，做这种拆分几乎等于重写。

当然，我这个量级完全没有拆微服务的必要。单服务 + 分层架构 + 良好的配置管理，已经足以应对日常需求。微服务是手段不是目的，这个认知要时刻保持。

6.3 几个实在的建议

最后分享几个我在这次重构中总结出来的实操建议，不算什么大道理，但能帮你少走弯路：

• 先留着旧脚本，不要删。我重构过程中不止一次想放弃，新代码跑不通时，回退到旧脚本先顶着是最省心的方案。等新服务稳定运行一两周再删旧的。
• 接口范围做小，逻辑深度做全。与其开 20 个接口每个都是半吊子，不如先开 3 个接口但把异常处理、日志、健康检查都做扎实。
• 测试用例跟着核心逻辑走。不一定要求全覆盖，但数据清洗、指标计算这种"纯计算"逻辑必须有测试。以后改代码不心虚。
• 写文档。哪怕只是 README 里写清楚启动命令、环境变量、目录结构，三个月后的你会感谢现在的你。

这次脚本改造服务的过程，本质上是一次从"能用"到"可控"的升级。技术栈只是手段，真正重要的是把职责理清、把边界定好、把运行状态暴露出来。做完之后你会发现，写代码的心态都变了——以前是在维护一个"石敢当"，碰一下就得拜一拜；现在是在维护一个透明、稳定、可扩展的系统，改起来有底气多了。