FastAPI构建LLM SaaS的Auth与文件上传核心骨架

1. 项目概述：为什么一个LLM SaaS的起点，必须是“能登录、能传文件”的骨架

做LLM应用开发的朋友，大概率都踩过这个坑：兴致勃勃搭好大模型推理接口，本地调通了，连上前端也跑起来了，结果一上线就卡在第一步——用户从哪来？文件往哪存？权限怎么管？我去年带团队落地三个垂直领域LLM SaaS产品，前两个项目都在“认证+上传”环节反复返工：第一个项目用自研JWT逻辑，两周后发现刷新令牌没做滑动窗口，被客户投诉会话频繁中断；第二个项目直接套用Django Allauth，结果整个用户表结构和OAuth2流程跟FastAPI生态不兼容，硬改了四天SQL迁移脚本。直到第三个项目，我们才真正把“Auth and File Upload”拆出来，作为独立可复用、可审计、可压测的模板层。它不是功能模块，而是SaaS系统的呼吸系统——没有它，再炫的推理链路也只是离线Demo。这个模板解决的不是“能不能做”，而是“能不能稳、能不能审、能不能扩”。它面向三类人：刚从Notebook转向生产环境的算法工程师，需要零配置接入企业级身份体系；独立开发者想快速验证PMF（Product-Market Fit），但又不想在登录页上花三天；还有技术负责人，要确保上传路径符合GDPR/等保2.0对文件元数据、访问日志、生命周期的强制要求。核心关键词——FastAPI、LLM SaaS、Auth、File Upload——不是并列关系，而是因果链：Auth是文件上传的前提，文件上传是LLM服务的输入入口，而FastAPI是让这条链路在高并发下不掉链子的底层承重墙。

2. 整体设计思路：为什么放弃“全栈框架”，选择“协议级解耦”

2.1 不选Django或Flask的底层逻辑

很多人第一反应是：“直接用Django Admin配个用户模型，再加个django-storages不就完事了？”我试过。在内部POC阶段，Django确实省了两天时间，但当客户提出“必须对接我们AD域的LDAP，并且所有登录失败要实时推送到SIEM平台”时，问题来了：Django的认证中间件是单点强耦合的，你得重写authenticate()方法，还要绕过它的Session机制去注入自定义审计钩子。更麻烦的是文件上传——Django默认走FileSystemStorage，但客户要求所有上传文件必须加密落盘，且密钥由HashiCorp Vault动态分发。这时候你会发现，框架自带的save()方法根本没法插手加密流。而FastAPI的哲学是“协议即契约”：它不预设你用什么数据库、什么存储、什么认证协议，只强制你遵守OpenAPI规范和ASGI生命周期。我们把Auth拆成三个协议层：

• 传输层：强制HTTPS + HSTS头，所有密码字段用SecretStr封装，避免日志泄露；
• 协议层：OAuth2 Password Flow + Bearer Token，拒绝任何Cookie-based Session，为后续JWT无状态扩展留口；
• 实现层：用passlib做bcrypt哈希（不是argon2，因为客户审计要求明确写明“需支持FIPS 140-2验证的哈希算法”，而bcrypt有NIST认证版本）；

文件上传同理。我们不用UploadFile直接存磁盘，而是抽象出FileStorageBackend接口，当前实现是S3CompatibleStorage（兼容MinIO、Cloudflare R2、阿里云OSS），但预留了EncryptedLocalStorage和IPFSBackend的继承点。这种设计让客户说“下周要切到私有IPFS集群”，我们只需要重写一个类，改两行DI注入，不用碰任何路由或业务逻辑。

2.2 为什么Auth必须前置到Lifespan事件中

FastAPI的lifespan事件常被当成“启动时初始化DB连接”的地方，但我们把它用成了Auth系统的“心跳检测中枢”。模板里有段关键代码：

@app.on_event("startup") async def startup_event(): # 1. 预热JWT密钥缓存（避免首请求解密延迟） await jwt_manager.preload_keys() # 2. 检查S3桶策略是否允许POST（防止部署后上传500） await file_storage.validate_write_permission() # 3. 启动后台任务：每5分钟扫描过期临时上传文件 asyncio.create_task(cleanup_expired_uploads())

这段逻辑解决了真实生产中的三个痛点：

• 首屏加载慢：JWT解密需要RSA公钥，如果每次请求都从磁盘读取PEM文件，P95延迟会突增80ms。我们提前加载进内存，并用cachetools.TTLCache缓存72小时；
• 配置漂移：S3桶策略可能被运维手动修改，导致上传接口静默失败。validate_write_permission()会尝试发一个预签名POST请求，失败则直接sys.exit(1)，让K8s健康检查立刻拉起新Pod；
• 磁盘爆满：用户上传10GB文件后关掉页面，临时文件不会自动清理。我们用aiocron定时扫描/tmp/uploads/下超过2小时未被file_upload_session表引用的文件——注意，这里不是删文件，而是先写入cleanup_log表（含文件名、SHA256、操作人ID），再执行删除，满足审计溯源要求。

这种把Auth的“可用性保障”下沉到启动阶段的设计，让整个SaaS系统从第一行日志开始就处于可审计、可监控状态，而不是等用户报错才被动响应。

2.3 文件上传的“双通道”架构：为什么不用单一multipart/form-data

LLM SaaS的文件上传有两类典型场景：

• 小文件高频：用户上传PDF合同（<5MB），每秒3-5次，要求首字节响应<200ms；
• 大文件低频：用户上传10GB医学影像DICOM序列，要求断点续传、分片校验、上传进度可查；

如果只用FastAPI原生UploadFile，会遇到硬伤：

• UploadFile.file.read()是同步阻塞IO，在高并发下会吃光Event Loop线程；
• 没有分片能力，大文件上传超时后只能重传全部，浪费带宽；
• 无法获取客户端真实IP（Nginx反代后只有X-Forwarded-For，但UploadFile不暴露headers）。

我们的解法是“双通道”：

• 通道A（默认）：POST /v1/upload接收multipart/form-data，适用于<10MB文件。底层用anyio.to_thread.run_sync()将shutil.copyfileobj()扔进线程池，避免阻塞Event Loop；
• 通道B（大文件专用）：POST /v1/upload/init返回预签名URL（如https://bucket.s3.amazonaws.com/uploads/{uuid}/part-{n}?X-Amz-Signature=...），客户端直传S3，服务端只收POST /v1/upload/complete触发合并逻辑。

关键细节在于：两个通道最终都写入同一张file_metadata表，字段完全一致（id,original_name,size_bytes,sha256,uploader_id,status），只是storage_path字段值不同（通道A存/tmp/uploads/xxx.pdf，通道B存s3://bucket/uploads/xxx/）。这样下游LLM服务调用get_file_content(file_id)时，完全感知不到差异——它只管查表、按storage_path协议去读，彻底解耦上传方式与消费方式。

3. 核心细节解析：Auth与文件上传的12个实操陷阱与避坑方案

3.1 JWT Token的“三重过期”设计：为什么不能只靠exp字段

很多教程教你在create_access_token()里写expires_delta=timedelta(hours=1)，这在SaaS里是危险操作。我们实际采用“三重过期”机制：

过期类型	字段位置	触发条件	处理方式
Token级过期	JWT payloadexp	时间到达	前端自动跳转登录页
Session级过期	Redis键session:{user_id}:{jti}	用户主动登出/管理员踢出	DELETE该键，下次请求verify_token()时因jti不存在而拒绝
Refresh级过期	JWT payloadrefresh_exp	刷新令牌超时（通常7天）	强制重新登录

提示：jti（JWT ID）不是UUID，而是sha256(f"{user_id}_{timestamp}_{random_str}")，避免Redis键名被枚举。我们用redis-py的pipeline批量执行SET session:{uid}:{jti} 1 EX 604800和SET user:{uid}:current_jti {jti} EX 604800，保证原子性。

实操中最大的坑是“时钟漂移”。客户服务器时间比NTP源快3分钟，导致Token明明没过期却被拒。解决方案：在verify_token()里加入leeway=120参数（python-jose库支持），允许最多2分钟偏差。但更根本的是——我们在K8s Deployment里强制挂载hostPath: /etc/timezone，并用initContainer运行ntpd -q -p pool.ntp.org校时，从基础设施层掐断漂移源头。

3.2 密码重置的“时间窗+次数锁”双重防护

标准的“发送重置邮件→点击链接→填新密码”流程，在SaaS里必须防暴力枚举。我们的实现包含三个硬性约束：

1. 时间窗限制：每个邮箱24小时内最多触发3次重置请求。用Redis Hash存储reset_attempts:{email}，key为{timestamp}，value为1，过期时间设为86400秒。每次请求前HLEN reset_attempts:{email}，超限返回429 Too Many Requests；
2. 链接时效性：重置Token有效期严格为15分钟，且jti绑定用户ID+时间戳哈希，用完即焚；
3. 密码强度审计：新密码必须通过zxcvbn库检测（不是正则！），要求熵值≥30bit。例如Tr0ub4dour&3会被判弱密码，因为它是常见单词变形；而correct-horse-battery-staple熵值高达44bit，直接放行。

注意：zxcvbn的Python版zxcvbn-python有严重性能问题（单次检测耗时200ms+），我们改用Rust重写的zxcvbn_rs，耗时压到8ms内。这是实测下来唯一能在P99延迟<50ms要求下可用的方案。

3.3 文件上传的“元数据分离”实践：为什么不能把文件名存在数据库里

新手常犯错误：INSERT INTO files (name, size, path) VALUES (?, ?, ?)。这在多语言环境下必崩——用户上传简历_张三.pdf，MySQL如果用utf8mb3编码，张三会变成??。更糟的是安全风险：../../../etc/passwd这种路径遍历，如果直接拼接进open()函数，就是RCE漏洞。

我们的解法是“元数据分离”：

• 原始文件名：存入original_filename字段，类型为VARCHAR(512)，字符集utf8mb4；
• 安全文件名：生成secure_filename = f"{uuid4().hex}_{int(time.time())}_{hashlib.md5(original_filename.encode()).hexdigest()[:8]}.bin"，只存这个；
• 存储路径：固定为/uploads/{year}/{month}/{day}/{secure_filename}，由file_storage类统一管理；

关键代码片段：

def generate_secure_filename(original: str) -> str: # 移除所有控制字符和路径分隔符 clean = re.sub(r'[\x00-\x1f\x7f/\\:*?"<>|]', '_', original) # 截断过长文件名（Windows限制255字符） if len(clean) > 200: name, ext = os.path.splitext(clean) clean = f"{name[:180]}{ext}" # 添加时间戳和哈希防重名 timestamp = int(time.time()) hash_part = hashlib.md5(original.encode()).hexdigest()[:8] return f"{uuid4().hex}_{timestamp}_{hash_part}{os.path.splitext(clean)[1] or '.bin'}"

实测效果：上传../../etc/shadow→ 存为a1b2c3d4_1712345678_1a2b3c4d.bin，原始名在数据库里完整保留，但物理路径绝对安全。

3.4 OAuth2第三方登录的“域白名单”强制校验

客户要求“只允许公司邮箱登录”，但OAuth2 Provider（如Google）返回的email字段可被伪造。我们的防护链有三层：

1. Provider端配置：在Google Cloud Console里设置Authorized domains为yourcompany.com，这是最外层防火墙；
2. Token校验层：verify_google_token()函数不仅验签名，还检查hd（hosted domain）claim是否存在且等于yourcompany.com；
3. 数据库约束层：users.email字段加唯一索引，但类型为VARCHAR(254) COLLATE utf8mb4_0900_as_cs（大小写敏感），防止[e-mail]和[e-mail]被当成两个用户。

实操心得：Google的hdclaim在使用G Suite账号时才返回，普通gmail账号没有。所以必须在OAuth2 Flow里强制添加hd=yourcompany.com参数，否则state参数会被忽略。我们用fastapi-users的OAuth2AuthorizeParams类封装此逻辑，避免每个Provider重复写。

3.5 大文件上传的“分片校验”实现：如何避免网络抖动导致数据损坏

当客户端直传S3时，分片上传（Multipart Upload）是标配，但默认的ETag校验不可靠——S3对每个Part返回的ETag是MD5哈希，但对合并后的Object，ETag可能是"abc123..."（非MD5）。我们强制要求客户端在每个Part上传前计算sha256(part_bytes)，并在POST /v1/upload/complete时提交所有Part的SHA256列表。服务端收到后：

1. 调用S3.list_parts()获取所有已上传Part的ETag；
2. 对每个Part发起HEAD请求，读取自定义Headerx-amz-meta-sha256（客户端上传时设置）；
3. 比对客户端提交的SHA256与S3返回的x-amz-meta-sha256是否一致；
4. 全部一致才执行complete_multipart_upload()，否则返回400 Bad Request并提示具体哪个Part校验失败。

这个设计让网络抖动导致的Part损坏能被精准定位，而不是等到LLM服务读取时才发现UnicodeDecodeError。实测中，某客户CDN节点故障导致3%的Part上传损坏，此机制让问题在10秒内被发现并重传，而非等待2小时后用户反馈“文档解析失败”。

3.6 Auth中间件的“无感降级”策略：当Redis宕机时怎么办

Redis是Auth的核心依赖，但如果它挂了，整个SaaS不能瘫痪。我们的中间件auth_dependency实现了三级降级：

async def auth_dependency( request: Request, token: str = Depends(oauth2_scheme), ) -> User: try: # Level 1: 正常走Redis校验jti payload = jwt_manager.decode(token) if not await redis_client.exists(f"session:{payload['sub']}:{payload['jti']}"): raise HTTPException(401, "Invalid session") return await get_user_by_id(payload["sub"]) except RedisConnectionError: # Level 2: 降级为JWT签名+exp校验（无状态） payload = jwt_manager.decode_no_redis(token) return await get_user_by_id(payload["sub"]) except Exception as e: # Level 3: 最终兜底——记录告警，但放行（仅限内部测试环境） if settings.ENV == "staging": logger.warning(f"Auth fallback triggered: {e}") return await get_user_by_id(1) # 默认admin raise e

关键经验：decode_no_redis()函数不查jti，但强制校验iss（issuer）、aud（audience）和exp，并用pydantic做payload schema校验。这保证即使Redis全挂，用户仍能登录，只是无法被管理员踢出——符合“可用性优先于强一致性”的SaaS设计原则。

4. 实操过程详解：从零搭建可交付的Auth+Upload模板

4.1 环境准备与依赖锁定：为什么用Poetry而不选pip-tools

项目根目录的pyproject.toml关键片段：

[tool.poetry.dependencies] python = "^3.11" fastapi = {version = "^0.110.0", extras = ["all"]} sqlalchemy = {version = "^2.0.29", extras = ["mypy"]} alembic = "^1.13.1" redis = "^4.6.0" boto3 = "^1.28.50" passlib = {version = "^1.7.4", extras = ["bcrypt"]} python-jose = {version = "^3.3.0", extras = ["cryptography"]} zxcvbn-rs = "^0.12.0"

选择Poetry的核心原因是依赖冲突可视化。比如python-jose[cryptography]和cryptography本身有版本交叉约束，pip install会静默降级，而Poetry的poetry lock --no-update会明确报错：“cryptography>=39.0.0required bypython-jose, butcryptography==38.0.4locked”。我们实测过，用pip-tools生成的requirements.txt在CI流水线里有7%概率因网络波动导致cryptography安装失败，而Poetry的poetry export -f requirements.txt生成的锁文件100%可重现。

实操步骤：

1. poetry init创建基础配置；
2. poetry add fastapi sqlalchemy alembic逐个添加主依赖；
3. poetry add --group dev pytest pytest-asyncio httpx添加测试依赖；
4. poetry lock生成poetry.lock；
5. poetry export -f requirements.txt --without-hashes > requirements.txt供Docker构建使用。

特别注意：--without-hashes是必须的，因为Docker构建时pip install -r requirements.txt如果带hash，会因镜像内时区差异导致pydantic校验失败（hash计算依赖datetime.now()）。

4.2 数据库迁移：为什么用Alembic而不手写SQL

alembic revision --autogenerate -m "add users and files tables"生成的versions/xxx_add_users_and_files_tables.py内容如下：

def upgrade(engine): op.create_table( "users", sa.Column("id", sa.Integer(), primary_key=True), sa.Column("email", sa.String(254), nullable=False, unique=True), sa.Column("hashed_password", sa.String(128), nullable=False), sa.Column("is_active", sa.Boolean(), default=True), sa.Column("created_at", sa.DateTime(), default=sa.func.now()), sa.Column("updated_at", sa.DateTime(), default=sa.func.now(), onupdate=sa.func.now()), ) op.create_table( "file_metadata", sa.Column("id", sa.Integer(), primary_key=True), sa.Column("original_filename", sa.String(512), nullable=False), sa.Column("secure_filename", sa.String(255), nullable=False, unique=True), sa.Column("size_bytes", sa.BigInteger(), nullable=False), sa.Column("sha256", sa.String(64), nullable=False), sa.Column("uploader_id", sa.Integer(), sa.ForeignKey("users.id"), nullable=False), sa.Column("status", sa.Enum(FileStatus), default=FileStatus.UPLOADED), sa.Column("created_at", sa.DateTime(), default=sa.func.now()), ) # 关键：添加索引提升查询性能 op.create_index("ix_file_metadata_uploader_id", "file_metadata", ["uploader_id"]) op.create_index("ix_file_metadata_status", "file_metadata", ["status"])

为什么必须加索引？实测数据：当file_metadata表有50万条记录时，SELECT * FROM file_metadata WHERE uploader_id=123不加索引耗时2.3秒，加索引后压到12ms。而LLM服务每次处理请求都要查用户最近上传的3个文件，这个查询是QPS最高的SQL之一。

迁移执行命令：

• 本地开发：alembic upgrade head；
• 生产部署：在K8s Job里执行alembic upgrade ${MIGRATION_VERSION}，用initContainer确保DB连接成功后再运行；
• 回滚：alembic downgrade -1（只允许回退一步，强制人工审核）。

注意：alembic revision --autogenerate会漏掉Enum类型变更，必须手动在upgrade()里补op.execute("ALTER TYPE file_status ADD VALUE 'PROCESSING'")，否则PostgreSQL会报错。

4.3 Auth路由实现：从/login到/me的完整链路

auth/router.py核心代码：

@router.post("/login") async def login( form_data: OAuth2PasswordRequestForm = Depends(), db: AsyncSession = Depends(get_db), ) -> TokenResponse: user = await authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException(400, "Incorrect email or password") if not user.is_active: raise HTTPException(400, "Inactive user") # 生成Access Token和Refresh Token access_token_expires = timedelta(hours=1) refresh_token_expires = timedelta(days=7) access_token = create_access_token( data={"sub": user.id, "jti": str(uuid4())}, expires_delta=access_token_expires, ) refresh_token = create_refresh_token( data={"sub": user.id, "jti": str(uuid4())}, expires_delta=refresh_token_expires, ) # 将refresh_token jti存入Redis await redis_client.setex( f"refresh:{user.id}:{refresh_token_jti}", int(refresh_token_expires.total_seconds()), "1" ) return TokenResponse( access_token=access_token, token_type="bearer", expires_in=int(access_token_expires.total_seconds()), refresh_token=refresh_token, ) @router.get("/me", response_model=UserRead) async def read_users_me(current_user: User = Depends(current_active_user)): return current_user

关键细节：

• OAuth2PasswordRequestForm自动从x-www-form-urlencoded解析username/password，无需手动request.form()；
• create_access_token()和create_refresh_token()是两个独立函数，因为Refresh Token的scope应更小（只含refresh），且jti必须不同；
• current_active_user依赖项里，get_current_user()先解Token，再查DB，最后校验is_active，三层校验缺一不可；

实测陷阱：OAuth2PasswordRequestForm默认不支持application/json，如果前端用fetch()发JSON，会返回422 Unprocessable Entity。解决方案是在main.py里加全局异常处理器：

@app.exception_handler(RequestValidationError) async def validation_exception_handler(request, exc): if "application/json" in request.headers.get("content-type", ""): # 尝试从JSON body提取username/password try: body = await request.json() if "username" in body and "password" in body: # 重写form_data逻辑... except: pass return JSONResponse({"detail": "Invalid request"}, status_code=422)

4.4 文件上传路由：支持multipart和pre-signed URL双模式

upload/router.py核心实现：

@router.post("/upload", response_model=FileUploadResponse) async def upload_file( file: UploadFile = File(...), current_user: User = Depends(current_active_user), db: AsyncSession = Depends(get_db), file_storage: FileStorageBackend = Depends(get_file_storage), ) -> FileUploadResponse: # 1. 校验文件大小（前端JS校验可被绕过） if file.size > settings.MAX_UPLOAD_SIZE: raise HTTPException(413, f"File too large. Max size: {settings.MAX_UPLOAD_SIZE}") # 2. 生成安全文件名 secure_name = generate_secure_filename(file.filename) # 3. 保存到存储后端 storage_path = await file_storage.save( file=file, secure_filename=secure_name, user_id=current_user.id, ) # 4. 写入数据库 file_meta = FileMetadata( original_filename=file.filename, secure_filename=secure_name, size_bytes=file.size, sha256=await calculate_sha256(file), uploader_id=current_user.id, storage_path=storage_path, ) db.add(file_meta) await db.commit() await db.refresh(file_meta) return FileUploadResponse( id=file_meta.id, original_filename=file.filename, size_bytes=file.size, uploaded_at=file_meta.created_at, ) @router.post("/upload/init", response_model=PreSignedUrlResponse) async def init_multipart_upload( filename: str = Form(...), current_user: User = Depends(current_active_user), file_storage: FileStorageBackend = Depends(get_file_storage), ) -> PreSignedUrlResponse: # 生成唯一upload_id upload_id = str(uuid4()) # 获取预签名URL（S3兼容） presigned_url = await file_storage.generate_presigned_post( filename=filename, upload_id=upload_id, user_id=current_user.id, ) return PreSignedUrlResponse( upload_id=upload_id, presigned_url=presigned_url, fields=presigned_url.fields, # 包含x-amz-signature等 )

关键参数说明：

• settings.MAX_UPLOAD_SIZE = 10 * 1024 * 1024（10MB），这是multipart通道的硬上限；
• generate_presigned_post()返回的fields包含key,x-amz-algorithm,x-amz-credential,x-amz-date,x-amz-signature，客户端需用FormData.append()逐个设置；
• upload_id不存数据库，只作为S3key前缀（如s3://bucket/uploads/{upload_id}/part-1），避免元数据膨胀。

实测中，某客户前端用axios发multipart，但没设Content-Type: multipart/form-data; boundary=xxx，导致FastAPI解析失败。解决方案：在main.py加中间件自动修复：

@app.middleware("http") async def fix_multipart_header(request: Request, call_next): if request.method == "POST" and "multipart/form-data" in request.headers.get("content-type", ""): # 强制设置boundary（即使header里没写） if "boundary=" not in request.headers["content-type"]: new_boundary = f"----WebKitFormBoundary{uuid4().hex}" request._headers = MutableHeaders(request.headers) request._headers["content-type"] = f"multipart/form-data; boundary={new_boundary}" response = await call_next(request) return response

4.5 测试覆盖：为什么用HTTPX而不选TestClient

tests/test_auth.py示例：

@pytest.mark.asyncio async def test_login_success(client: AsyncClient, test_user: User): response = await client.post( "/auth/login", data={"username": test_user.email, "password": "testpassword"}, ) assert response.status_code == 200 data = response.json() assert "access_token" in data assert data["token_type"] == "bearer" # 验证token能解码 payload = jwt.decode(data["access_token"], settings.JWT_SECRET_KEY, algorithms=["HS256"]) assert payload["sub"] == test_user.id @pytest.mark.asyncio async def test_upload_file(client: AsyncClient, test_user: User, auth_header: dict): # 创建测试文件 test_file = b"Hello, LLM SaaS!" response = await client.post( "/upload/upload", files={"file": ("test.txt", test_file, "text/plain")}, headers=auth_header, ) assert response.status_code == 200 data = response.json() assert data["original_filename"] == "test.txt" assert data["size_bytes"] == len(test_file)

选择httpx.AsyncClient而非TestClient的原因：

• TestClient是同步的，无法测试async def路由的真实行为；
• httpx.AsyncClient能真实模拟HTTP/2、流式响应、超时重试等网络行为；
• 可以用pytest-asyncio标记，与AsyncSession测试无缝集成。

实操技巧：auth_headerfixture这样写：

@pytest.fixture def auth_header(test_user: User) -> dict: token = create_access_token(data={"sub": test_user.id}) return {"Authorization": f"Bearer {token}"}

避免在每个test里重复生成token，提升测试速度。

5. 常见问题与排查技巧实录：来自12个生产环境的真实战报

5.1 问题速查表：高频故障与根因定位

现象	日志线索	根因	解决方案
401 Unauthorized频发	JWT decode error: Signature verification failed	Nginx代理转发时修改了Authorizationheader大小写（如authorization）	在Nginx配置中加proxy_pass_request_headers on;，并确保underscores_in_headers on;
文件上传后500 Internal Server Error	OSError: [Errno 28] No space left on device	/tmp分区满（Docker容器默认/tmp只有64MB）	在Dockerfile里加RUN mkdir -p /data/uploads && chmod 777 /data/uploads，settings.UPLOAD_DIR = "/data/uploads"
422 Unprocessable Entityon/login	detail: [{"loc":["body"],"msg":"field required","type":"value_error.missing"}]	前端发JSON但后端期待x-www-form-urlencoded	检查前端Content-Type，或按4.3节加全局异常处理器
403 Forbiddenon S3 pre-signed URL	The request signature we calculated does not match the signature you provided.	客户端上传时Content-Type与预签名时声明的不一致	在generate_presigned_post()里显式指定Fields={"Content-Type": "binary/octet-stream"}，客户端必须匹配
504 Gateway Timeouton large file upload	Nginx error log:upstream timed out (110: Connection timed out)	Nginxproxy_read_timeout默认60秒，大文件上传超时	在Nginx配置中加proxy_read_timeout 3600;（1小时）

5.2 Redis连接池泄漏的隐蔽征兆与修复

某次上线后，Redis连接数从20飙升到1024，INFO clients显示connected_clients:1024，但client list里只有20个活跃连接。根因是redis-py的连接池未正确关闭。我们在lifespan里这样修复：

@app.on_event("shutdown") async def shutdown_event(): # 显式关闭Redis连接池 if hasattr(redis_client, "connection_pool"): await redis_client.connection_pool.disconnect() # 关闭数据库连接池 await db_engine.dispose()

关键经验：redis-py的disconnect()必须await，否则是异步任务，可能被Event Loop丢弃。我们用pytest的monkeypatch测试此逻辑：

def test_shutdown_disconnects_redis(monkeypatch): mock_pool = Mock() mock_pool.disconnect = AsyncMock() monkeypatch.setattr(redis_client, "connection_pool", mock_pool) # 触发shutdown asyncio.run(shutdown_event()) mock_pool.disconnect.assert_called_once()

5.3 PostgreSQL死锁的现场还原与预防

死锁日志示例：

ERROR: deadlock detected DETAIL: Process 12345 waits for ShareLock on transaction 67890; blocked by process 67890. Process 67890 waits for ShareLock on transaction 12345; blocked by process 12345.

根因是两个事务以不同顺序更新users和file_metadata表。我们的预防措施：

• 统一更新顺序：所有事务先UPDATE users再UPDATE file_metadata，绝不颠倒；
• 减少事务粒度：/upload/complete不再在一个事务里完成“更新file_metadata状态+插入processing_job”，而是拆成两个独立API；
• 加锁提示：在SELECT ... FOR UPDATE时加SKIP LOCKED，避免长事务阻塞。