先说说我为什么要折腾这件事。之前有一个跑了大半年的 Python 脚本每天凌晨从几个数据源拉取内容、清洗、算指标、把结果写进数据库。功能一直没出过大事故但说实话它离让人放心差得很远。我判断脚本是否正常的唯一方式是去看最终产出的数据文件是不是新鲜的。有一天数据文件是空的我甚至分不清是昨天凌晨就挂了还是今天刚坏——脚本里全是散落的printWindows 计划任务里的输出早找不到了。那次之后我下定决心要把这个脚本重构成一个真正的服务。最终落地的是一个基于 FastAPI 的常驻进程按照 API 层、Service 层、Repository 层的分层架构来组织。这篇文章把整个思考过程和实操细节完整写下来为什么选 FastAPI 而不是 Flask 或 Django、目录结构怎么定、旧脚本的逻辑怎么拆、部署时踩了哪些坑——尤其是 uvicorn 日志丢失那个经典问题。如果你手上也有一堆能跑但不可控的脚本正打算把它们服务化这篇文章应该能给你一些直接的参考。1. 脚本能跑就行我先说说为什么非要改成服务1.1 脚本时代的三个致命短板很多人觉得脚本只要能出结果就行没必要搞成什么服务。这个观点在自己用、低频、出错了看一眼就明白的阶段是成立的。但我的实际情况是脚本已经被多个业务方间接依赖了数据文件一空下游报表、接口、其他人的分析全都会受影响而我只能在出事之后被动发现。我把脚本时代的痛点归纳成三个无观测脚本跑完就退出中间过程只剩print。在计划任务/后台任务里这些输出要么丢失、要么被吞掉出了问题只能盲猜。无并发脚本是单进程串行的。一旦有两个调用方同时触发要么加锁、要么互相踩数据怎么处理都别扭。无边界所有逻辑耦合在一个main()和几个辅助函数里。表面上看改动方便实际上任何一处变化都会引发连锁反应改完还得担心会不会影响其他环节。用一个生活类比来说脚本就像你家里的一个抽屉钥匙、零钱、水电费单全放一起。平时拿东西确实方便但有一天你想把水电费单单独交给另一个人去处理就会发现所有东西纠缠在一起根本没法单独分出去。脚本里的函数也一样数据获取、清洗、计算、落库全挤在一个函数里看着没毛病可一旦有了复用交接扩展这些需求就开始痛苦了。1.2 出现这几种信号说明改造时机到了不是所有脚本都需要改造成服务。我给自己总结了几个必须改造的信号你可以对照一下脚本开始被别人调用——这里的别人可能是另一个程序、另一台机器、或者一个网页按钮。同一套逻辑在多个脚本里重复出现——获取数据、清洗、落库这些代码拷贝来拷贝去。需要对外提供查询能力——比如同事问你今天的数据跑出来了吗结果是多少你只能去翻数据库没法给他一个入口。答不上来现在状态怎么样的问题——脚本当前在跑哪个步骤、上次成功是什么时候、这次失败卡在哪一行你完全没有概念。一旦出现两条以上就别再往脚本里打补丁了。我当时同时中了四条所以彻底下决心重写。1.3 为什么选 FastAPI而不是 Flask 或 Django选型这件事我认真比过不是跟风。当时候选有 Flask、Django、FastAPI 三个各自特点差异很大。框架优点在我这个场景里的问题Flask简单、生态成熟、上手快太自由了没有强制的结构约束很容易从一个脚本写成一个更大的泥球Django全家桶、自带 ORM 和 Admin太重。脚本改服务核心需求是把逻辑暴露成接口Django 的很多能力用不上反而增加了一堆概念要学FastAPIPydantic 校验、依赖注入、自动生成 OpenAPI 文档、原生 async几乎没有短板唯一的问题是社区相比前两者略年轻但核心功能足够稳定最终让我定下心来的是 FastAPI 的依赖注入和Pydantic 模型。依赖注入天然引导你拆分层——每个接口声明自己需要什么 Service而不是在业务函数里到处 new 依赖。这跟把脚本拆干净的目标高度一致。哪怕你的脚本逻辑比我的还复杂这个思路也适用。2. 重构前夜把旧脚本拆成一幅职责地图2.1 动刀之前先给旧脚本画图这一步很容易被跳过但恰恰是最关键的。我当时没有直接开写新代码而是先把旧脚本的每一个函数、每一个流程列出来按获取数据 / 处理数据 / 输出数据三个维度分组。比如我这个脚本本质上是三段式从外部数据源拉取原始内容HTTP 请求 解析清洗、去重、计算指标纯逻辑处理把结果写进数据库落库听起来很简单对不对但当你真的去翻代码时会发现这三件事互相穿插拉数据时顺带做了格式整理计算时又回头查了上一次的缓存落库时还顺手改了一下数据内容。这种穿插就是脚本时代最大的技术债。我的做法是拿张纸或者用一个文档把每个函数的输入、输出、副作用比如修改了哪个全局变量、写了哪个文件都标出来。这个动作花不了太长时间但后面写分层代码时你会省掉大量返工的痛苦。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_encodingutf-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_code400, 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] DescriptionData Service (FastAPI) Afternetwork.target [Service] Userwww-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 Restartalways RestartSec3 [Install] WantedBymulti-user.target几个细节EnvironmentFile用来加载环境变量文件这样配置不必写死在 systemd 配置里。Restartalways是必须的脚本时代最大的痛就是挂了不知道现在服务挂了会自动拉起。WorkingDirectory必须正确否则日志路径、配置相对路径全都对不上。激活命令sudo systemctl daemon-reload sudo systemctl enable>app.get(/health) def health_check(): # 这里可以检查数据库连接是否正常、外部依赖是否可用 return {status: alive}这个接口有两个用途一是部署平台的探活二是你自己快速确认服务状态不用去翻日志。真正生产环境我还会往里面塞一点依赖状态检查比如数据库能不能连、外部数据源通不通。这样负载均衡器或者监控系统就能在服务带病运行前及时知道。另外systemd 的Restartalways解决了崩溃重启的问题但优雅退出同样重要当服务收到终止信号时应该让正在处理的请求跑完而不是立刻被掐断。uvicorn 本身对 SIGTERM 的处理已经比较完善只要你不强制kill -9它会给正在处理的请求一个宽限期。真正需要注意的反而是你的业务代码别在进程退出逻辑里做太重的清理操作否则会拖慢重启速度。6. 重构之后的收益以及我会怎么继续演进6.1 三个立竿见影的改变重构完成并稳定运行之后有几个改变是立竿见影的。第一是可观测性。现在任何时间点我都能通过GET /api/v1/tasks/status知道任务在不在跑、上次结果怎么样。配合日志文件出问题时定位速度从小时级变成了分钟级。第二是可控性。之前脚本只在计划任务里定时跑想手动触发一次就要登录服务器敲命令。现在直接发一个请求就行甚至可以接一个简单的管理页面。数据源的地址变了也只需要改配置或调用配置接口不用改代码重新部署。第三是可测试性。分层之后每个 Service 方法都可以独立测。我用 pytest 写了针对核心业务逻辑的测试用例跑一遍只要几秒钟。脚本时代根本没有这种体验因为所有逻辑都耦合在一起想测一个函数就得先让前面的步骤全部跑一遍。6.2 从单体服务到微服务的演进空间这次分层架构还有一个隐藏收益它为将来拆微服务留好了路。比如后来我如果要把数据拉取和指标计算拆成两个独立服务Service 层和 Repository 层的边界已经天然把它俩分开了从接口调用变成 RPC/消息队列调用改动范围是可控的。如果你一开始就是平铺的脚本代码做这种拆分几乎等于重写。当然我这个量级完全没有拆微服务的必要。单服务 分层架构 良好的配置管理已经足以应对日常需求。微服务是手段不是目的这个认知要时刻保持。6.3 几个实在的建议最后分享几个我在这次重构中总结出来的实操建议不算什么大道理但能帮你少走弯路先留着旧脚本不要删。我重构过程中不止一次想放弃新代码跑不通时回退到旧脚本先顶着是最省心的方案。等新服务稳定运行一两周再删旧的。接口范围做小逻辑深度做全。与其开 20 个接口每个都是半吊子不如先开 3 个接口但把异常处理、日志、健康检查都做扎实。测试用例跟着核心逻辑走。不一定要求全覆盖但数据清洗、指标计算这种纯计算逻辑必须有测试。以后改代码不心虚。写文档。哪怕只是 README 里写清楚启动命令、环境变量、目录结构三个月后的你会感谢现在的你。这次脚本改造服务的过程本质上是一次从能用到可控的升级。技术栈只是手段真正重要的是把职责理清、把边界定好、把运行状态暴露出来。做完之后你会发现写代码的心态都变了——以前是在维护一个石敢当碰一下就得拜一拜现在是在维护一个透明、稳定、可扩展的系统改起来有底气多了。