PyUploadX 文件与目录上传服务

PyUploadX 是一个轻量、可独立部署、可水平扩展的文件与目录上传服务，提供统一的 文件上传、断点续传、目录上传、生命周期管理能力。详细设计与实现约束见 产品设计说明书（§31 目录结构、§16 REST API、§17 SDK）。

功能特性

• 上传模式：小文件 Proxy 上传、大文件 Multipart（分片 + 预签名）、Local/S3/MinIO 存储适配
• 断点续传：SDK 与 Portal 均支持指纹校验、状态恢复、缺失分片重传、URL 刷新
• 目录上传：Manifest + NDJSON、路径安全校验、.uploadignore、冲突策略、双层并发
• 生命周期：permanent / ttl / expires_at / temporary / sliding_ttl，Legal Hold、Webhook
• 集群一致性：无状态 API 节点、PostgreSQL 行锁、Part Upsert、Complete/Abort 幂等
• 可观测性：结构化 JSON 日志、Prometheus 指标、/healthz /readyz /startupz

技术栈

FastAPI · SQLAlchemy 2（ORM）· Alembic · PostgreSQL · Redis · boto3 · Python SDK（httpx）· React + TypeScript + Vite + Dexie · Docker Compose · Kubernetes · pytest

快速开始（Docker Compose）

PostgreSQL / Redis / MinIO 等第三方组件不随应用镜像构建：默认通过环境变量指向 本地或已有实例（默认地址 host.docker.internal，即宿主机；端口与凭据均可覆盖）。

方式 A：使用本地/已有 PostgreSQL、Redis、MinIO

# 可选：覆盖默认地址与凭据
export UPLOAD_DATABASE_URL='postgresql+asyncpg://upload:upload@localhost:5432/uploads'
export UPLOAD_REDIS_URL='redis://localhost:6379/0'
# 哨兵形态改为（连接面由公共工厂 ikc_sdk.redis 装配，与四个引擎的 <PREFIX>_* 口径一致）：
# export UPLOAD_REDIS_MODE=sentinel
# export UPLOAD_SENTINEL_NODES='10.0.0.1:26380,10.0.0.2:26380,10.0.0.3:26380'
# export UPLOAD_SENTINEL_MASTER_NAME=mymaster UPLOAD_SENTINEL_PASSWORD=<master 口令> UPLOAD_SENTINEL_DB=7
export UPLOAD_STORAGE__S3__INTERNAL_ENDPOINT_URL='http://localhost:9000'
export S3_ACCESS_KEY=minioadmin S3_SECRET_KEY=minioadmin
# 推荐：portal 自动获取 token（OpenResty 注入 X-API-Key），无需在浏览器粘贴 API Key。
# 设置 PORTAL_API_TOKEN 可用固定 token；不设置则每次启动生成随机 token。
bash scripts/start-stack.sh           # 等价于 docker compose up -d --build（额外下发 portal token）

Portal 打开后会自动完成鉴权（OpenResty 注入 token）；若部署环境未注入 token，则显示登录页， 输入有效的 API Key 后进入（Token 保存在 localStorage，刷新或重启浏览器后保持登录）。

Portal 主要界面：

• 缺省进入「文件浏览」：左侧按 Bucket/目录的导航树筛选，右侧表格分页展示，支持 下载、复制下载链接、删除、状态过滤、表头点击排序（服务端排序，跨页生效）；
• 「上传」页支持设置 Bucket、目标前缀与生命周期（永久 / 定时过期 / 指定时间过期， 含到期动作与 TTL 时长）。

方式 B：自带第三方组件（deploy/infra/compose.yaml，端口可用 POSTGRES_PORT、 REDIS_PORT、MINIO_PORT 覆盖）

docker compose -f deploy/infra/compose.yaml up -d          # postgres/redis/minio + 建桶
docker compose up -d --build                               # 应用服务

自带组件的数据通过命名卷外部持久化（postgres-data / minio-data），不进镜像； 删除 / 重建容器数据保留；备份直接针对卷或宿主目录操作。

单节点/集群一键（含组件）：

docker compose -f deploy/single-node/compose.yaml up -d --build
docker compose -f deploy/cluster/compose.yaml up -d --build --scale upload-api=3 --scale worker=2

运行架构：应用由同一代码库、两类镜像体系构建（Dockerfile target api → upload-api 与一次性 migrate，target worker → upload-worker；portal/Dockerfile → upload-portal）。 单节点固定单副本（显式 container_name）；集群模式使用 --scale 水平扩展无状态 API/Worker， 数据库迁移由 migrate 容器（alembic upgrade head）在 API 启动前完成。日志可选用 IKC Log Center 投递（log_center 配置段，默认关闭）。

一键构建全部项目镜像（upload-api / worker / portal / gateway / migrate / 加固 MinIO）：

bash scripts/build-images.sh            # 构建全部项目镜像
bash scripts/build-images.sh --export   # 构建并 docker save 导出到 docker/images/

加固 MinIO 镜像 pyuploadx-minio-haproxy:latest（随 IMAGE_PREFIX，ikc 口径为 ikc-pyuploadx-minio-haproxy:latest）让 MinIO 仅监听容器内回环地址、外部只走 HAProxy（9000 S3 / 9001 控制台）；数据必须外部挂载（-v /data/minio:/data），不进镜像。

启动后：

• API/OpenAPI：http://localhost:8000/docs
• Portal：http://localhost:5173（API Key：dev-key）
• MinIO Console：http://localhost:9001（minioadmin / minioadmin）
• 生产网关（可选）：pyuploadx-gateway（OpenResty HTTPS，TLS 终止 + 反代，部署见 docs/deploy-offline.md §12）

Kubernetes 部署

模板位于 deploy/kubernetes/（Job/migrate / Namespace / Deployment / Service / Ingress / HPA / PDB / Secret / ConfigMap / ServiceMonitor）。发布前先执行数据库迁移 （Job/upload-migrate，alembic upgrade head，一次成功即完成），再滚动发布 API/Worker。 API 默认 3 副本、maxUnavailable: 0、优雅终止 60s，并通过 /readyz 探针摘除故障节点。 ServiceMonitor（servicemonitor.yaml）可被 Prometheus Operator 直接发现：

kubectl apply -f deploy/kubernetes/migrate-job.yaml   # 先执行数据库迁移（一次性 Job）
kubectl apply -f deploy/kubernetes/

日志可选投递到 IKC Log Center：在 ConfigMap/upload-config 的 log_center 段配置 （默认关闭，enabled: false）。

生产强制 HTTPS：Ingress 终止 TLS；单节点/集群部署由 deploy/nginx/（OpenResty 网关， pyuploadx-gateway）强制跳转 HTTPS 并拒绝非白名单 Origin 的 CORS 请求。

本地开发

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,docs]"
python -m pytest tests -q          # 全量测试
ruff check app sdk upload_service tests scripts
uvicorn app.main:create_app --factory --reload --host 0.0.0.0 --port 8000

测试默认使用 SQLite（aiosqlite）；CI 使用 PostgreSQL。集群模式禁止 SQLite（config validate 会拒绝）。

数据库迁移（Alembic）

alembic upgrade head    # 升级到最新 schema
alembic downgrade -1    # 回退一个版本（仅限向后兼容的迁移）
make migrate            # 等价于 alembic upgrade head

Schema 变更一律通过 Alembic 迁移交付，禁止用临时脚本改表（见 AGENTS.md Architecture Contracts）。

MinIO / S3 Adapter 测试

UPLOAD_MINIO_TEST=1 \
UPLOAD_STORAGE__S3__INTERNAL_ENDPOINT_URL=http://localhost:9000 \
S3_ACCESS_KEY=minioadmin S3_SECRET_KEY=minioadmin \
python -m pytest tests/integration/test_s3_storage.py -q

未设置 UPLOAD_MINIO_TEST=1 时该套件自动跳过；CI 内置 MinIO 服务并始终运行它。

Portal E2E（Playwright）

cd portal
npm ci
npx playwright install chromium      # 首次运行需下载浏览器
npx playwright test                   # 自动拉起 uvicorn(8000) 与 vite(5173)

E2E 覆盖登录与错误展示、文件/目录上传、生命周期策略、刷新后队列恢复（§29.6）。 本地默认 E2E_API_URL=http://127.0.0.1:8000、E2E_PORTAL_URL=http://127.0.0.1:5173， API Key 为 e2e-key（由 portal/playwright.config.ts 注入，SQLite + local 存储）。 CI 的 portal job 会安装浏览器依赖并执行同一套测试。

配置

配置优先级：代码默认值 < config/config.yaml < 环境变量（UPLOAD_SECTION__FIELD）< 命令行。

python -m upload_service config validate
python -m upload_service config show --redact-secrets
python -m upload_service reconcile upload {upload_id} --dry-run

关键环境变量：

变量	说明
UPLOAD_API_KEYS	JSON：{"tenant/principal": ["key"]} 或 key 数组
UPLOAD_DATABASE_URL	SQLAlchemy 异步 URL
UPLOAD_REDIS_URL	Redis URL（single / replica 形态；要连哨兵代理就是这一形态 + 代理地址）
UPLOAD_REDIS_MODE	连接形态：single（缺省）/ replica / sentinel；UPLOAD_REDIS__ENABLED=false 可关闭
UPLOAD_SENTINEL_NODES	sentinel 形态必填：host:port,host:port,...；另有 UPLOAD_SENTINEL_MASTER_NAME（缺省 mymaster）、UPLOAD_SENTINEL_PASSWORD（master 口令）、UPLOAD_SENTINEL_AUTH_PASSWORD（哨兵节点口令，缺省复用 master 口令）、UPLOAD_SENTINEL_DB（缺省 0）
S3_ACCESS_KEY / S3_SECRET_KEY	S3/MinIO 凭据

备份与恢复

运维手册见 docs/operations.md：PostgreSQL 每日全量备份与 PITR、 对象存储 Versioning/复制/对象锁、恢复顺序（先 DB → 对象存储 → Redis → API 只读检查 → Reconcile Dry Run → Worker → 开放入口）、向后兼容的迁移与回滚原则（§28）。

Python SDK

完整 SDK 开发文档（方法参考、数据模型、异常映射、测试与发版）见 docs/docs_sdk-dev.md。

from datetime import timedelta
from pyuploadx import Client, FileLifecycle

client = Client(
    base_url="http://localhost:8000",
    api_key="dev-key",
    state_dir="~/.pyuploadx/uploads",
)

# 通用上传：目录 -> 目录上传；>= 8 MiB -> Multipart 分片 + 断点续传；否则 -> Proxy 小文件上传
result = client.upload(
    "./README.md",
    bucket="app-default",
    lifecycle=FileLifecycle.ttl(timedelta(days=30)),
)

job = client.upload(
    "./album-assets",
    bucket="app-default",
    destination_prefix="artists/10001/albums/2026",
    exclude=[".git/**", "**/*.tmp"],
    conflict_policy="reject",
)

# 下载：返回本地落盘 Path，直接输出文件名
saved = client.download(result.id, "/tmp/README.md")
print(saved, saved.name)                       # 本地路径与文件名

# 通过 URL 获取原始文件名（预签名 URL / 永久链接）
url = client.get_download_url(result.id, expires_seconds=3600)
filename = client.filename_from_url(url)
print(filename)                                # => README.md
client.download_from_url(url, f"/tmp/{filename}")

pyuploadx.Client 是 UploadClient 的通用扩展（上传 + 下载，原类未改动）： upload() 自动按路径选择小文件 / 大文件 / 目录上传策略，filename_from_url() 从下载 URL 解析 原始文件名。UploadClient 的全部方法与参数保持不变，旧代码兼容。

安装

pip install pyuploadx          # 官方 PyPI：Python SDK（Python ≥ 3.11，第三方依赖仅 httpx）
pip install pyuploadx-server   # 官方 PyPI：服务端（FastAPI，Docker 部署时通常无需 pip 安装）
pip install dist/pyuploadx-0.10.0-py3-none-any.whl  # 或仓库直装（历史版本清单见 dist/README.md）

SDK 与服务端为两个独立发布包（pyuploadx / pyuploadx-server）；发版与版本管理见 docs/docs_product-design.md §37。

获取结果与状态

上传接口同步返回结果对象；文件上传后可随时查询最新状态、生命周期并下载。

# 上传结果：FileInfo（id/status/etag/size_bytes/lifecycle_mode/expires_at）
# 响应自带临时预签名下载 URL（expires_in 秒；Local 后端为 None）
info = client.upload("./README.md", bucket="app-default")
print(info.id, info.status, info.size_bytes, info.original_filename, info.download_url, info.expires_in)

# 目录上传结果：DirectoryJobInfo（status/uploaded_files/failed_files/uploaded_bytes）
job = client.upload("./album-assets", bucket="app-default", destination_prefix="artists/10001")
print(job.status, job.uploaded_files, job.failed_files)

# 查询接口：服务端最新状态（文件 / 分片会话 / 目录任务）
info = client.get_file(info.id)                  # 最新 FileInfo
session = client.get_upload(session_id)          # 分片会话状态（UploadSessionInfo）
job = client.get_directory_job(job.id)           # 目录任务状态（DirectoryJobInfo）

# 生命周期查询与下载
lifecycle = client.get_lifecycle(info.id)
url = client.get_download_url(info.id, expires_seconds=3600)   # 过期后重新获取 URL
saved = client.download(info.id, "/tmp/downloaded.bin")        # 代理流式下载，返回落盘 Path
print(saved, saved.name, info.original_filename)               # 本地路径 / 本地文件名 / 服务端原始文件名
print(client.filename_from_url(url))                           # 通过 URL 获取原始文件名
client.download(info.id, "/tmp/big.bin", concurrency=8)  # 超大文件：并发 Range 分片
client.download_from_url(url, "/tmp/downloaded.bin")           # 直接下载 URL（预签名/永久链接）

上传策略、URL 下载与文件名

import httpx

# 1) 通用上传：Client.upload 自动分派；directory 自动拼接 object_key = <directory>/<文件名>
info = client.upload(
    "./report.pdf",
    bucket="app-default",
    directory="reports/2026",          # => object_key: reports/2026/report.pdf
)
large = client.upload(
    "./model.bin",
    bucket="app-default",
    directory="models/backup",         # => object_key: models/backup/model.bin
    part_size=8 * 1024 * 1024,
    concurrency=4,
)

# 2) 上传响应返回临时预签名下载 URL（expires_in 秒；Local 后端为 None）
print(info.download_url, info.expires_in)
url = client.get_download_url(info.id, expires_seconds=3600)   # 过期后重新获取

# 3) 通过 URL 获取原始文件名
filename = client.filename_from_url(url)                       # 预签名 URL -> 路径末段（URL 解码）
print(filename)                                                # => report.pdf

# 4) 使用 URL 下载：SDK 流式写盘（支持 progress 回调），并输出落盘文件名
saved = client.download_from_url(
    url, f"/tmp/{filename}",
    progress=lambda done, total: print(f"{done}/{total}"),
)
print(saved, saved.name)
# 等价：client.download(info.id, f"/tmp/{filename}", url=url)

# 原始 httpx 等价实现
with httpx.stream("GET", url, follow_redirects=True) as resp:
    resp.raise_for_status()
    with open(f"/tmp/{filename}", "wb") as f:
        for chunk in resp.iter_bytes():
            f.write(chunk)

REST API 摘要

GET  /healthz | /readyz | /startupz | /metrics
GET  /v1/files                 POST /v1/files/upload
GET/DELETE /v1/files/{id}
GET  /v1/files/{id}/download   POST /v1/files/{id}/presign-download
POST /v1/files/{id}/permanent-link   GET /v1/files/{id}/download-link   # 永久下载链接
POST /v1/uploads               POST /v1/uploads/resume
GET  /v1/uploads/{id}          GET /v1/uploads/{id}/parts
POST /v1/uploads/{id}/parts/presign
PUT  /v1/uploads/{id}/parts/{part_number}
POST /v1/uploads/{id}/parts/commit
POST /v1/uploads/{id}/refresh | /complete | /abort
POST /v1/directory-uploads ... /v1/files/{id}/lifecycle ... /v1/client-config

错误响应统一为 {"error": {"code", "message", "details", "retryable", "request_id"}}。

可观测性

• 健康检查：/healthz（Liveness）、/readyz（Readiness，检测数据库连通性）、/startupz。
• 指标：/metrics（Prometheus 文本格式），覆盖上传/分片/断点/目录/生命周期/数据库/Redis/ Storage 全链路；Kubernetes 下由 deploy/kubernetes/servicemonitor.yaml 抓取。
• 日志：结构化 JSON（request_id / trace_id / node_id / tenant_id / duration_ms 等）， 不记录 Secret、完整 API Key 或完整预签名 URL（§23）。

性能测试

python scripts/benchmark_upload.py --base-url http://localhost:8000 \
    --api-key dev-key --files 100 --concurrency 16
python scripts/benchmark_upload.py --large-mb 64 --part-size 8388608 --concurrency 8

场景对应 §29.7：并发小文件 Proxy 上传与 1 GiB 级大文件 Multipart（走 SDK 断点续传链路）， 输出吞吐与 p50/p95 延迟。大规模场景（10 万小文件、目录总大小 1 TiB）请在目标环境按需调参。

项目结构

app/        FastAPI 后端（api/config/core/db/storage/services/lifecycle/worker）
sdk/pyuploadx/  Python 客户端 SDK
portal/     React + TypeScript Portal（Dexie/IndexedDB 断点状态）
deploy/     单节点/集群 Compose、Kubernetes（含 ServiceMonitor）、OpenResty 网关、MinIO 引导
config/     YAML 配置示例
tests/      单元与集成测试
scripts/    渲染/文档检查/看板同步/性能测试脚本
docs/       设计文档、运维手册与架构图（SVG 源 / PNG 生成）

数据库与 ORM 契约

• 所有数据访问必须使用 SQLAlchemy 2 ORM（app/db/models.py + app/db/repositories/）。
• 禁止裸 SQL 字符串 DML；唯一例外是仓储层内基于 sqlalchemy.dialects.postgresql.insert 的 ON CONFLICT Upsert。
• Schema 变更通过 Alembic 迁移交付。

安全要点

• 生产强制 HTTPS（OpenResty/Ingress TLS 终止），CORS 仅允许显式配置的 Origin。
• Secret 只通过环境变量注入，禁止写入 YAML 或提交到仓库；日志脱敏密钥与完整签名 URL。
• PostgreSQL/Redis 不暴露公网（Compose 仅内部网络，K8s 无外部 Service）。
• 对象 Key 与目录相对路径均做防路径逃逸校验；上传会话校验所有权，Complete/Abort 幂等。

许可证

MIT