很多项目一开始都处于“无标题”状态——需求只有一句话，连名字都没想好。这篇文章记录的正是这样一个项目：从一句“我要一个能传文件、能处理文件的服务”开始，最终落地成一个基于 FastAPI 的轻量级文件中转服务。它支持文件上传、存储、异步压缩与格式转换，生成带有效期的临时下载链接，并自动清理过期文件。整个过程不依赖重型框架，也没有复杂的分布式组件，适合想从零搭后端服务的开发者和需要快速搭建内部工具的工程师参考。文中所有架构决策、参数计算、踩坑记录都来自实际运行经验，可以直接在你的项目里复用。
1. 项目定位与整体设计：为什么选“文件中转服务”
1.1 原始需求拆解：一个连名字都没有的项目到底要做什么
当初接手这个项目时，需求就这么一段话：用户要能上传文件，系统要能对文件做点处理，处理完之后给一个链接让别人下载。没有说并发量，没有说文件类型，甚至没说要不要做用户系统。
这种模糊需求最容易犯的错就是上来就写代码。我的习惯是先不问“怎么做”，而是把所有场景拉出来列一遍，看看这个服务实际要承担哪些职责：
- 用户要从浏览器或命令行上传文件，单个文件大小需要有限制；
- 文件要落盘保存，不能因为上传后服务重启就丢；
- 对图片要能压缩，对其他常见格式要做简单转换；
- 处理逻辑可能耗时，上传接口不能因为处理文件而阻塞；
- 生成一个只在一定时间内有效的下载链接；
- 过期文件要自动清理，不能把磁盘撑爆。
拆完这些，项目的形状才开始清晰：它本质上是一个“文件中转站”，不负责长期保管数据，只负责在有限时间内完成“上传—处理—分享”这条链路。最后我给它起了一个临时工作名：轻量文件处置服务。这个定位直接影响后续所有技术选型——单机部署、轻量依赖、快速开发。
1.2 技术栈选型：Python + FastAPI 的理由和取舍
选型之前我先定了几个硬性条件：开发时间要短、部署要简单、代码量要可控。基于这三点，我直接排除了重框架和复杂中间件。
后端框架选择上，我对比过 Flask 和 FastAPI。Flask 足够简单，但它在异步处理上不够原生，文件上传和后台任务需要额外引入库来配合，文档生成也很弱。FastAPI 自带 OpenAPI 文档，上传接口、参数校验、依赖注入都是开箱即用，而且基于 asyncio 构建异步能力很方便。异步对这个项目非常关键，因为文件处理是典型的 I/O 密集操作，能异步化就异步化。
数据库选型很多人在这一步容易上头。对于这个项目，数据量最多几千条记录，单机部署，没有多写并发，SQLite 完全够用。用 PostgreSQL 不是不行，但会引入连接管理和备份成本，对一个小工具来说是过度设计。
任务队列也做了取舍。Celery 是成熟方案，但它必须配一个消息中间件，比如 Redis 或 RabbitMQ，部署成本立刻上来了。我最终选择直接在 FastAPI 应用内部用asyncio.Queue实现一个轻量级任务队列，配合几个固定数量的 worker 协程来消费任务。这个方案在单进程、任务量几百上千的场景下完全可靠，而且少一个中间件就少一个故障点。
取舍原则很简单：项目复杂度决定了技术复杂度。个人工具和内部系统，优先选择“能少装一个组件就少装一个组件”。
1.3 服务架构总览：上传、存储、处理、清理四个环节
整个服务的核心链路是四个环节，我已经在需求阶段把它们拆清楚了，对应到架构上就是四块职责：
上传环节：接收文件流和元数据，校验文件类型与大小，将文件流落盘。 存储环节：保存文件到本地磁盘，同时在数据库里记录文件的原始名、存储名、大小、状态、过期时间这些元信息。 处理环节：任务 worker 从队列中取出任务，执行压缩、格式转换等操作，完成后更新数据库中的任务状态和文件状态。 清理环节：定时任务扫描数据库中的过期记录，删除物理文件并清理对应数据行。
请求走向是这样的：客户端 POST 一个文件到/upload，接口校验通过后先把文件写入临时目录，然后向任务队列里投放一个处理任务，最后立即返回一个 task_id。客户端拿着 task_id 查询状态，等状态变成 DONE 之后，就可以拼接出下载链接。
我没在这套架构里加消息队列，也没做独立 worker 进程，因为初步评估下来单机单进程的压力很小。如果后续量上来，这个架构的演进路径也很明确：把asyncio.Queue替换成 Redis Stream，把 worker 独立成单独进程，其他环节都不需要动。
2. 核心模块拆解：接口设计、存储策略与异步处理
2.1 上传接口设计：大小限制、分片与断点续传的取舍
上传接口是整个服务的门面，我把限制条件和接口字段先定清楚，避免后续被动。
文件大小默认限制在 100MB。这个数值不是拍脑袋定的，而是基于实际使用场景算出来的——这个服务主要用于分享日志、导出的数据文件、压缩包和图片，单个文件基本都在 10MB 以内。100MB 留出了足够余量，同时又能避免超大文件撑爆临时目录。生产环境的 Nginx 也需要同步调整client_max_body_size，这块我在部署部分细说。
文件类型校验上，我选择白名单策略。允许的类型是：图片类（jpg、png、webp、gif）、文档类（pdf、txt、csv）、压缩包类（zip、tar.gz）。白名单的明显好处是缩小了可执行文件上传的风险面。
关于分片上传和断点续传，我直接放弃了。原因有两个：第一，分片上传需要前后端配合做切片合并，实现成本至少在两天以上，对这个体量的工具不划算；第二，100MB 以内的文件在正常网络下，单次 multipart 上传已经足够快，断点续传这个功能换来的体验提升很有限。如果哪天需要传大文件，再引入基于流式上传的分片方案也不迟。
接口字段设计上，只保留了三个：文件本身、一个可选的说明备注、一个可选的处理参数（比如图片压缩质量）。代码里用 FastAPI 的UploadFile和File依赖来接收。
from fastapi import FastAPI, UploadFile, File, Form from pydantic import BaseModel app = FastAPI() ALLOWED_EXTENSIONS = {"jpg", "jpeg", "png", "webp", "gif", "pdf", "txt", "csv", "zip"} MAX_SIZE = 100 * 1024 * 1024 # 100MB @app.post("/upload") async def upload_file( file: UploadFile = File(...), note: str = Form(default=""), quality: int = Form(default=80), ): # 校验扩展名 ext = file.filename.rsplit(".", 1)[-1].lower() if ext not in ALLOWED_EXTENSIONS: return {"code": 1, "msg": "文件类型不允许"} # 校验大小 size = 0 content = await file.read() size = len(content) if size > MAX_SIZE: return {"code": 2, "msg": "文件超过大小限制"} # 后续处理放到存储服务里 task_id = storage.save_uploaded_file(file.filename, content, ext, note, quality) return {"code": 0, "task_id": task_id}这里有个经验值得单独说：不要用file.size这个属性来判断大小，因为在某些实现里它可能为空。实际读取完文件流再计算长度更稳妥。同时，一次性read()对整个文件内容进内存，对 100MB 上限来说是可控的，后续优化方向是改成流式写入磁盘，避免内存峰值。
2.2 文件名处理与存储结构：为什么哈希目录比按日期更稳
原始文件名绝对不能直接拿来落盘，这是文件服务的第一条安全底线。用户上传的文件名可能包含中文、空格、特殊字符，甚至有路径穿越的尝试，比如../../etc/passwd。直接把原始名拼进路径里就是在给攻击者送漏洞。
我的方案是：存储名一律用 UUID 生成，保留原文件的后缀名，数据库里的原始文件名单独存一列，只在用户下载时作为响应头里的Content-Disposition文件名使用。
import uuid import os import time def generate_storage_name(original_name: str) -> str: ext = original_name.rsplit(".", 1)[-1].lower() return f"{uuid.uuid4().hex}.{ext}" def generate_storage_path() -> str: # 目录按月分，方便清理 month_dir = time.strftime("%Y%m") return os.path.join("uploads", month_dir)目录结构我做了两层划分：外层按月份（202506、202507），内层按 UUID 前两个字符做子目录。这样做的好处有两个：第一，单目录文件数不会膨胀，查找文件时磁盘压力小；第二，月初清理上个月文件时直接删整个目录，效率极高。
有人习惯用文件名哈希值来做路径分拆，uuid.hex的前两位天然均匀分布，效果类似但更简单——不需要额外计算哈希，文件名本身就是随机值。
数据库表设计是这个模块的核心，字段尽可能一次到位。我用 SQLite，建表语句直接写在服务初始化时执行。
CREATE TABLE IF NOT EXISTS files ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT NOT NULL UNIQUE, original_name TEXT NOT NULL, storage_name TEXT NOT NULL, storage_path TEXT NOT NULL, file_size INTEGER NOT NULL, file_ext TEXT NOT NULL, note TEXT DEFAULT '', status TEXT DEFAULT 'PENDING', -- PENDING/RUNNING/DONE/FAILED created_at INTEGER NOT NULL, expire_at INTEGER NOT NULL );task_id用uuid4生成，它既是对外的任务查询凭证，也是下载链接的一部分。expire_at存的是过期时间戳，默认生成时加 7 天。查询时只需要看一下expire_at < 当前时间就知道该不该清理。
2.3 异步任务处理：压缩、格式转换这类耗时操作怎么不阻塞请求
文件上传接口返回要快，但图片压缩、格式转换这些操作的时间是不可控的——一张 20MB 的图片压缩可能要几秒甚至几十秒，用户不可能一直等接口返回。所以任务必须异步化。
方案对比再做一次：Celery 太重，需要中间件；BackgroundTasks是 FastAPI 自带的后台任务机制，但它只是协程池的简单封装，任务状态不好追踪，也没有队列积压的概念。我选择的asyncio.Queue + worker 协程方案，代码量小、行为可控。
具体实现是两段逻辑：上传接口往队列里投放任务，后台启动固定数量的 worker 协程并发消费。
import asyncio from typing import Dict, Any task_queue = asyncio.Queue() task_status: Dict[str, Any] = {} # 内存中的任务状态表 async def worker(worker_id: int): while True: task = await task_queue.get() task_id = task["task_id"] try: task_status[task_id] = "RUNNING" # 执行具体处理任务 await process_file(task) task_status[task_id] = "DONE" except Exception as e: task_status[task_id] = "FAILED" print(f"worker {worker_id} task {task_id} failed: {e}") finally: task_queue.task_done() async def process_file(task: Dict[str, Any]): # 按文件扩展名分发到不同处理器 ext = task["file_ext"] if ext in ("jpg", "jpeg", "png", "webp", "gif"): await compress_image(task) else: # 非图片文件直接标记完成 pass # 启动 4 个 worker @app.on_event("startup") async def startup(): for i in range(4): asyncio.create_task(worker(i))这里要注意一个细节：task_status放在内存里意味着服务重启后状态就丢了。对内部工具可以接受，但对生产环境需要把状态持久化到数据库。我实际是把状态写进files表的status字段，查询接口直接查数据库，不查内存，这样至少状态不丢。
worker 数量选 4 是我在测试机上实测出来的。2 核 4G 的机器上，图片压缩是 CPU 密集操作，开 8 个 worker 反而会因为上下文切换降低吞吐。4 个 worker 时总耗时和资源占用是最均衡的。
3. 实操落地：从零搭建这个服务的完整过程
3.1 环境准备与项目骨架
项目用 Python 3.11，依赖精简到极致。requirements.txt内容如下：
fastapi==0.111.0 uvicorn==0.30.1 python-multipart==0.0.9 pillow==10.3.0 aiosqlite==0.20.0python-multipart是 FastAPI 处理表单文件上传的必需依赖，漏装的话UploadFile会直接报错。aiosqlite用于异步读写 SQLite，避免同步数据库操作阻塞事件循环。
项目目录结构我习惯按职责分模块，而不是一个文件堆到底：
file-tool/ ├── app/ │ ├── main.py # 应用入口，挂载路由、启动 worker │ ├── routers/ │ │ ├── upload.py # 上传接口 │ │ ├── task.py # 任务状态查询接口 │ │ └── download.py # 下载接口 │ ├── services/ │ │ ├── storage.py # 文件落盘与元数据读写 │ │ ├── task.py # 队列与 worker 逻辑 │ │ ├── image.py # 图片压缩处理 │ │ └── cleanup.py # 定时清理任务 │ └── config.py # 配置常量 ├── uploads/ # 文件存储目录 ├── requirements.txt └── README.md装完依赖后，先创建uploads目录和app/config.py，把路径、过期时间、大小限制这些配置统一放在一处，后续调整不用翻代码。
3.2 核心代码实现：上传、任务查询、下载三个关键接口
上传接口我在前面已经写了核心片段，完整版需要调用存储服务把文件流写盘、生成存储路径、计算过期时间，然后入库并投放任务。我直接给出services/storage.py的关键实现。
import os import uuid import time import aiosqlite from . import config DB_PATH = os.path.join(config.BASE_DIR, "files.db") UPLOAD_DIR = config.UPLOAD_DIR EXPIRE_DAYS = config.EXPIRE_DAYS # 默认7天 async def save_uploaded_file(original_name: str, content: bytes, ext: str, note: str) -> str: task_id = uuid.uuid4().hex storage_name = f"{uuid.uuid4().hex}.{ext}" month_dir = time.strftime("%Y%m") sub_dir = time.strftime("%Y%m") + "/" + storage_name[:2] # 这里演示简单结构，生产按需分两层 target_dir = os.path.join(UPLOAD_DIR, month_dir, storage_name[:2]) os.makedirs(target_dir, exist_ok=True) target_path = os.path.join(target_dir, storage_name) with open(target_path, "wb") as f: f.write(content) expire_at = int(time.time()) + EXPIRE_DAYS * 86400 created_at = int(time.time()) async with aiosqlite.connect(DB_PATH) as db: await db.execute( "INSERT INTO files (task_id, original_name, storage_name, storage_path, file_size, file_ext, note, status, created_at, expire_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'PENDING', ?, ?)", (task_id, original_name, storage_name, target_path, len(content), ext, note, created_at, expire_at), ) await db.commit() return task_id查询接口逻辑简单，根据task_id查询数据库中的状态记录，但注意要对不存在或过期的记录区分返回码，不能用同一个错误信息含糊处理。
@app.get("/task/{task_id}") async def get_task_status(task_id: str): async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( "SELECT status, file_size, expire_at FROM files WHERE task_id = ?", (task_id,), ) row = await cursor.fetchone() if row is None: return {"code": 404, "msg": "任务不存在"} status, size, expire_at = row if expire_at < int(time.time()): return {"code": 410, "msg": "文件已过期"} return {"code": 0, "status": status, "size": size}下载接口的关键点在于：拼接物理文件路径时用数据库里的storage_path，绝不能拿原始文件名拼路径。下载文件名要设置filename*来处理中文，不然浏览器会乱码。
from fastapi.responses import FileResponse @app.get("/download/{task_id}") async def download_file(task_id: str): async with aiosqlite.connect(DB_PATH) as db: cursor = await db.execute( "SELECT storage_path, original_name FROM files WHERE task_id = ? AND status = 'DONE'", (task_id,), ) row = await cursor.fetchone() if row is None: return {"code": 404, "msg": "文件不存在或未处理完成"} storage_path, original_name = row if not os.path.exists(storage_path): return {"code": 404, "msg": "物理文件缺失"} headers = {"Content-Disposition": f'attachment; filename*=UTF-8\'\'{original_name}'} return FileResponse(storage_path, headers=headers)下载接口只允许状态为DONE的记录，这能避免文件还在处理中就被人下载到半成品。
3.3 部署配置与参数计算：并发数、超时时间、磁盘预警
部署我用了最常规的三件套：Nginx 反向代理 + Uvicorn 跑 FastAPI 应用 + 定时任务做清理。
Nginx 配置里最关键的几个参数是请求体大小、超时时间。请求体大小必须大于应用层的 100MB 限制，否则 Nginx 会在应用收到请求之前直接返回 413。我设置为 120MB，留出 multipart 表单的额外字节空间。超时时间上，proxy_read_timeout设置为 60 秒，因为上传接口本身不执行处理任务，60 秒足够完成一个 100MB 文件的上传。
server { listen 80; server_name file.example.internal; client_max_body_size 120M; proxy_read_timeout 60s; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }Uvicorn 启动命令要显式指定进程数和 worker 数量，我按照机器规格计算：机器是 2 核 4G，Uvicorn 的workers参数设置为 2，每个进程内部再启动 4 个任务 worker。总并发任务数是 2 * 4 = 8。测试时我用 50 张 5MB 图片并发上传压缩，8 个并发任务能在 30 秒内全部完成，这个数据完全可以接受。
磁盘预警必须做，不然迟早出事。我在/upload接口里加入前置检查：磁盘使用率超过 85% 时直接拒绝新上传，返回明确的错误码，避免服务在磁盘写满时报一堆难以排查的异常。
import shutil def check_disk_space(): usage = shutil.disk_usage(config.UPLOAD_DIR) used_percent = usage.used / usage.total * 100 return used_percent # 在 upload_file 入口处： if check_disk_space() > 85: return {"code": 3, "msg": "磁盘空间不足，请稍后再试"}清理逻辑用系统crontab每小时跑一次脚本，扫描过期记录，物理删除文件后清掉数据库记录。脚本不依赖应用自身机制，避免应用挂掉后清理任务也没了。
0 * * * * cd /opt/file-tool && python -m app.services.cleanupcleanup.py里的核心逻辑是分批删除，每批只处理 100 条记录。直接一次性删几千条记录会导致 SQLite 锁写，影响正在上传的请求。
4. 上线后的坑与排查：几个真实遇到过的问题
4.1 文件名中文乱码与路径冲突
这个项目上线第一天就遇到了中文文件名乱码问题。浏览器上传测试报告.pdf，下载时拿到的文件名变成了%E6%B5%8B%E8%AF%95...一长串编码。
排查过程很简单，打印出original_name发现系统读取到的就是 URL 编码后的字符串，问题出在浏览器端和 multipart 解析的编码处理上。后面我在下载接口里统一用filename*=UTF-8''格式把原始文件名放进去，浏览器看到这个响应头会自动解码成正确的中文文件名。这里有个细节，filename*后面的文件名一定要做 URL 编码，比如用urllib.parse.quote，不能直接把中文拼进去。
路径冲突的问题出现在 UUID 生成阶段。我一开始用uuid.uuid4().hex的前 8 位作为存储名，结果出现概率极低的碰撞恰好被测试撞上了。后来改成完整 32 位不截断，碰撞概率降到可忽略级别。
4.2 任务积压与 Worker 崩溃
上线第二周遇到过一次任务积压。现象是上传接口正常返回 task_id，但任务状态一直停在 PENDING，等了十分钟都没被消费。
查日志发现其中一个 worker 协程在处理一个超大图片时抛了异常，异常没有被task_done()保护住，导致队列消费循环提前退出。修复方案是在 worker 里做两层防护：外层 try/except 捕获所有异常，内层针对单任务的失败做记录而不是让整个 worker 退出。同时给任务处理加了超时控制，单任务处理超过 30 秒就强制标记为 FAILED，避免一个超大文件卡住整个队列。
这个经历让我养成了一个习惯：所有后台循环逻辑，不能只处理正常流程，必须把退出场景覆盖掉。worker 的while True里面，任何一个未经捕获的异常都会让整个消费链停摆。
4.3 临时文件清理策略与磁盘告警阈值
磁盘被撑爆这个坑是我自己挖的。最初清理任务只保留 3 天有效期的文件，但测试阶段大量上传的测试文件都把过期时间设成了默认 7 天，加上一次实验导入了几个 80MB 的压缩包，磁盘占用量直接冲到了 92%。
当时紧急处理流程是：手动停掉应用进程，用 SQL 找出所有expire_at已过期的记录，先删数据再删物理文件。这之后我做了两个调整：第一，在check_disk_space里把预警阈值从 85% 改为 80%，预留更多缓冲；第二，清理脚本增加一个硬保护逻辑，当磁盘使用率超过 90% 时，优先删除多天前已过期未清理的文件，并打印告警日志。
排查这类问题有一条通用路径：先看数据库记录是否还在，再看物理文件是否存在，两者对照才能定位是哪个环节出了错。很多文件丢失问题其实不是存储故障，而是清理任务误删了未过期的记录。
4.4 接口幂等与重复上报的验证技巧
最后再分享一个小技巧，是我在这个项目后续迭代中加上的。因为内网工具经常有脚本重试机制，同一个文件可能被上传好几次，产生多个任务和重复磁盘占用。我加了基于文件内容哈希的幂等校验：上传时计算一个 SHA-256 的头部向量，如果数据库中已经存在相同大小且相同前缀哈希的文件，直接返回原任务 ID，不再重复存储。这对日志和报告这类重复性高的文件效果特别明显，磁盘空间占用直接降了约三成。
这个项目跑到现在有三个多月，从“无标题”到稳定运行，最大体会是：再模糊的需求，只要能拆成“上传—存储—处理—清理”这样的清晰环节，后面所有细节都会自然长出来。技术选型上没有玄学，每一步取舍都是基于“这台机器只有 2 核 4G、用户只有内部几十号人”这类现实约束做出的。如果你也要搭类似服务，架构可以套用，但参数和限制条件一定要根据自己的实际场景重新算一遍。
