FastAPI Storage Module Skill
面向已有 FastAPI 项目的开发者，快速集成静态资源管理能力。
定位
- 目标：在已有
fastapi-init-skill骨架上，添加可运行的静态资源管理模块。 - 不替代：不重复生成
fastapi-init-skill已经提供的统一响应、JWT、SQLModel 等基础设施。 - 输出：模型、仓储、服务、路由、数据库迁移、接口契约、接入指南。
骨架依赖
本模块是
fastapi-init-skill的子模块，必须在骨架基础上使用。
使用前必须满足：
- ✅ 已安装
fastapi-init-skill（项目骨架） - ✅ 骨架包含：JWT、统一响应、SQLModel、分页、目录结构
检测逻辑：
- 读取用户项目根目录
SKILL.md或README.md - 检测是否包含
fastapi-init-skill相关内容 - 如未检测到骨架，提示："本模块需要先安装 fastapi-init-skill 骨架"
用户问题（最多 3 个）
1. 现有项目的包名是什么？（默认从骨架推断，如 app） 2. 表前缀是什么？（默认 wg） 3. 默认存储方式是什么？（local / aliyun / tencent / minio，默认 local）
核心能力清单
| # | 能力 | 说明 |
|---|---|---|
| 1 | 小文件上传 | 单文件/多文件直接上传（< 10MB） |
| 2 | 大文件切割上传 | 分片上传 + 断点续传（≥ 10MB） |
| 3 | 文件压缩 | 可选开启，支持 Pillow 压缩图片 |
| 4 | 文件下载 | 流式下载、断点续传下载 |
| 5 | 文件预览 | 图片/视频/文档在线预览 |
| 6 | 存储策略 | 本地存储 / 阿里云 OSS / 腾讯云 COS / MinIO |
| 7 | 附件管理 | 附件 CRUD、分类、标签 |
| 8 | 图片处理 | 缩略图、水印、格式转换 |
配置结构
python# config.py from pydantic import BaseSettings from typing import List, Optional class ChunkConfig(BaseSettings): enabled: bool = True size: int = 5 * 1024 * 1024 # 5MB 每片 threshold: int = 10 * 1024 * 1024 # 10MB 触发分片上传 class CompressConfig(BaseSettings): enabled: bool = False types: List[str] = ["jpg", "jpeg", "png"] quality: float = 0.8 max_size: int = 1024 * 1024 # 1MB 以上才压缩 class LocalConfig(BaseSettings): path: str = "./uploads" domain: str = "http://localhost:8000" class AliyunConfig(BaseSettings): access_key: str = "" secret_key: str = "" bucket: str = "" endpoint: str = "oss-cn-hangzhou.aliyuncs.com" class TencentConfig(BaseSettings): secret_id: str = "" secret_key: str = "" bucket: str = "" region: str = "ap-guangzhou" class MinioConfig(BaseSettings): endpoint: str = "http://localhost:9000" access_key: str = "minioadmin" secret_key: str = "minioadmin" bucket: str = "uploads" class StorageConfig(BaseSettings): type: str = "local" # local / aliyun / tencent / minio max_size: int = 100 * 1024 * 1024 # 100MB allowed_types: List[str] = [ "jpg", "jpeg", "png", "gif", "pdf", "doc", "docx", "xls", "xlsx", "zip", "mp4", "mp3" ] chunk: ChunkConfig = ChunkConfig() compress: CompressConfig = CompressConfig() local: LocalConfig = LocalConfig() aliyun: AliyunConfig = AliyunConfig() tencent: TencentConfig = TencentConfig() minio: MinioConfig = MinioConfig() storage_config = StorageConfig()
模块结构
src/storage/ ├── __init__.py ├── config.py # 配置 ├── constants.py # 常量 ├── exceptions.py # 异常 ├── models.py # SQLModel 模型 ├── schemas.py # Pydantic schemas ├── routers/ │ ├── __init__.py │ ├── upload.py # 上传接口 │ ├── download.py # 下载接口 │ └── attachment.py # 附件管理接口 └── services/ ├── __init__.py ├── storage_service.py # 存储策略接口 ├── local_storage.py # 本地存储 ├── oss_storage.py # 阿里云 OSS ├── cos_storage.py # 腾讯云 COS ├── minio_storage.py # MinIO ├── upload_service.py # 上传服务 ├── download_service.py # 下载服务 ├── attachment_service.py # 附件服务 └── image_service.py # 图片处理服务 alembic/versions/storage_module.py # 迁移文件 api-contract-storage.md # 接口契约 docs/storage-module-guide.md # 接入指南
核心实现
存储策略接口
python# services/storage_service.py from abc import ABC, abstractmethod from typing import BinaryIO, Optional class StorageService(ABC): """存储策略抽象接口""" @abstractmethod async def upload(self, file: BinaryIO, key: str, content_type: str) -> str: """上传文件 Args: file: 文件流 key: 存储 key content_type: 内容类型 Returns: 访问 URL """ pass @abstractmethod async def delete(self, key: str) -> None: """删除文件""" pass @abstractmethod async def get_url(self, key: str) -> str: """获取文件 URL""" pass @abstractmethod async def download(self, key: str) -> BinaryIO: """获取文件流""" pass
本地存储实现
python# services/local_storage.py import os import aiofiles from pathlib import Path from datetime import datetime from .storage_service import StorageService from ..config import storage_config class LocalStorageService(StorageService): """本地存储实现""" def __init__(self): self.base_path = Path(storage_config.local.path) self.domain = storage_config.local.domain self.base_path.mkdir(parents=True, exist_ok=True) async def upload(self, file: BinaryIO, key: str, content_type: str) -> str: """上传文件到本地""" file_path = self.base_path / key file_path.parent.mkdir(parents=True, exist_ok=True) async with aiofiles.open(file_path, 'wb') as f: content = await file.read() await f.write(content) return f"{self.domain}/{key}" async def delete(self, key: str) -> None: """删除本地文件""" file_path = self.base_path / key if file_path.exists(): file_path.unlink() async def get_url(self, key: str) -> str: """获取文件 URL""" return f"{self.domain}/{key}" async def download(self, key: str) -> BinaryIO: """获取文件流""" file_path = self.base_path / key if not file_path.exists(): raise FileNotFoundError(f"文件不存在: {key}") return open(file_path, 'rb')
上传服务（核心）
python# services/upload_service.py import uuid import math from datetime import datetime from typing import List, Optional, BinaryIO from fastapi import UploadFile from sqlmodel import Session, select from ..config import storage_config from ..models import Attachment, ChunkInfo from ..schemas import UploadResult, ChunkUploadInit from ..exceptions import StorageException from .storage_service import StorageService from .local_storage import LocalStorageService from .image_service import ImageService class UploadService: """上传服务""" def __init__(self, session: Session): self.session = session self.storage: StorageService = self._get_storage_service() self.image_service = ImageService() def _get_storage_service(self) -> StorageService: """获取存储服务实例""" storage_type = storage_config.type if storage_type == "local": return LocalStorageService() # elif storage_type == "aliyun": # return OssStorageService() # elif storage_type == "tencent": # return CosStorageService() # elif storage_type == "minio": # return MinioStorageService() else: raise StorageException(f"不支持的存储类型: {storage_type}") async def upload(self, file: UploadFile, category: str = "default") -> UploadResult: """上传文件（自动判断小文件/大文件）""" # 验证文件 await self._validate_file(file) # 判断是否需要分片上传 if self._need_chunk_upload(file.size): return await self._chunk_upload(file, category) # 小文件直接上传 return await self._direct_upload(file, category) async def _direct_upload(self, file: UploadFile, category: str) -> UploadResult: """小文件直接上传""" key = self._generate_key(file.filename) content_type = file.content_type or "application/octet-stream" # 是否需要压缩 content = await file.read() if self._need_compress(file.filename, len(content)): content = await self.image_service.compress( content, storage_config.compress.quality ) # 上传 from io import BytesIO url = await self.storage.upload(BytesIO(content), key, content_type) # 保存附件记录 attachment = Attachment( filename=file.filename, file_path=key, file_url=url, file_size=len(content), content_type=content_type, file_ext=self._get_extension(file.filename), category=category, storage_type=storage_config.type, is_compressed=self._need_compress(file.filename, file.size) ) self.session.add(attachment) self.session.commit() self.session.refresh(attachment) return UploadResult( id=attachment.id, url=url, filename=file.filename, size=len(content), content_type=content_type ) async def _chunk_upload(self, file: UploadFile, category: str) -> UploadResult: """大文件分片上传""" # 1. 初始化分片上传 upload_id = str(uuid.uuid4()) chunk_size = storage_config.chunk.size total_chunks = math.ceil(file.size / chunk_size) # 2. 保存分片信息 chunk_info = ChunkInfo( upload_id=upload_id, filename=file.filename, file_size=file.size, chunk_size=chunk_size, total_chunks=total_chunks, status=0 # 上传中 ) self.session.add(chunk_info) self.session.commit() # 3. 分片上传 for chunk_index in range(total_chunks): content = await file.read(chunk_size) chunk_key = f"chunks/{upload_id}/{chunk_index}" from io import BytesIO await self.storage.upload( BytesIO(content), chunk_key, "application/octet-stream" ) # 4. 合并分片 key = self._generate_key(file.filename) await self._merge_chunks(upload_id, key, total_chunks) # 5. 更新状态 chunk_info.status = 1 # 已完成 self.session.commit() # 6. 保存附件记录 url = await self.storage.get_url(key) attachment = Attachment( filename=file.filename, file_path=key, file_url=url, file_size=file.size, content_type=file.content_type, file_ext=self._get_extension(file.filename), category=category, storage_type=storage_config.type, is_compressed=False ) self.session.add(attachment) self.session.commit() self.session.refresh(attachment) return UploadResult( id=attachment.id, url=url, filename=file.filename, size=file.size, content_type=file.content_type ) async def get_uploaded_chunks(self, upload_id: str) -> List[int]: """获取已上传分片（断点续传）""" chunk_info = self.session.exec( select(ChunkInfo).where(ChunkInfo.upload_id == upload_id) ).first() if not chunk_info: raise StorageException("上传任务不存在") uploaded = [] for i in range(chunk_info.total_chunks): chunk_key = f"chunks/{upload_id}/{i}" try: await self.storage.download(chunk_key) uploaded.append(i) except Exception: pass return uploaded async def upload_chunk(self, upload_id: str, chunk_index: int, file: UploadFile): """上传单个分片（断点续传）""" chunk_info = self.session.exec( select(ChunkInfo).where(ChunkInfo.upload_id == upload_id) ).first() if not chunk_info: raise StorageException("上传任务不存在") chunk_key = f"chunks/{upload_id}/{chunk_index}" content = await file.read() from io import BytesIO await self.storage.upload(BytesIO(content), chunk_key, "application/octet-stream") def _need_chunk_upload(self, file_size: int) -> bool: """判断是否需要分片上传""" return (storage_config.chunk.enabled and file_size >= storage_config.chunk.threshold) def _need_compress(self, filename: str, file_size: int) -> bool: """判断是否需要压缩""" if not storage_config.compress.enabled: return False if file_size < storage_config.compress.max_size: return False ext = self._get_extension(filename) return ext.lower() in storage_config.compress.types def _generate_key(self, filename: str) -> str: """生成存储 key""" ext = self._get_extension(filename) date_path = datetime.now().strftime("%Y/%m/%d") return f"uploads/{date_path}/{uuid.uuid4()}.{ext}" def _get_extension(self, filename: str) -> str: """获取文件扩展名""" return filename.rsplit(".", 1)[-1] if "." in filename else "" async def _validate_file(self, file: UploadFile): """验证文件""" if not file.filename: raise StorageException("文件不能为空") if file.size > storage_config.max_size: raise StorageException("文件大小超过限制") ext = self._get_extension(file.filename) if ext.lower() not in storage_config.allowed_types: raise StorageException("不支持的文件类型")
上传路由
python# routers/upload.py from typing import List, Optional from fastapi import APIRouter, UploadFile, File, Form, Depends from sqlmodel import Session from ..schemas import UploadResult, ChunkUploadInit from ..services.upload_service import UploadService from ...core.deps import get_session router = APIRouter(prefix="/api/storage", tags=["存储管理"]) @router.post("/upload", response_model=UploadResult) async def upload_file( file: UploadFile = File(...), category: str = Form(default="default"), session: Session = Depends(get_session) ): """上传文件（自动判断小文件/大文件）""" service = UploadService(session) return await service.upload(file, category) @router.post("/upload/chunk/init", response_model=ChunkUploadInit) async def init_chunk_upload( filename: str = Form(...), file_size: int = Form(...), session: Session = Depends(get_session) ): """初始化分片上传""" service = UploadService(session) return await service.init_chunk_upload(filename, file_size) @router.post("/upload/chunk/{upload_id}/{chunk_index}") async def upload_chunk( upload_id: str, chunk_index: int, file: UploadFile = File(...), session: Session = Depends(get_session) ): """上传分片""" service = UploadService(session) await service.upload_chunk(upload_id, chunk_index, file) return {"message": "分片上传成功"} @router.post("/upload/chunk/{upload_id}/merge", response_model=UploadResult) async def merge_chunks( upload_id: str, filename: str = Form(...), category: str = Form(default="default"), session: Session = Depends(get_session) ): """合并分片""" service = UploadService(session) return await service.merge_chunks(upload_id, filename, category) @router.get("/upload/chunk/{upload_id}/chunks", response_model=List[int]) async def get_uploaded_chunks( upload_id: str, session: Session = Depends(get_session) ): """获取已上传分片（断点续传）""" service = UploadService(session) return await service.get_uploaded_chunks(upload_id)
下载路由
python# routers/download.py from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from sqlmodel import Session from ..services.download_service import DownloadService from ...core.deps import get_session router = APIRouter(prefix="/api/storage", tags=["存储管理"]) @router.get("/download/{attachment_id}") async def download_file( attachment_id: int, session: Session = Depends(get_session) ): """下载文件""" service = DownloadService(session) file_stream, filename, content_type = await service.download(attachment_id) return StreamingResponse( file_stream, media_type=content_type, headers={ "Content-Disposition": f'attachment; filename="{filename}"' } ) @router.get("/preview/{attachment_id}") async def preview_file( attachment_id: int, session: Session = Depends(get_session) ): """预览文件""" service = DownloadService(session) file_stream, filename, content_type = await service.preview(attachment_id) return StreamingResponse( file_stream, media_type=content_type, headers={ "Content-Disposition": f'inline; filename="{filename}"' } )
数据库模型
python# models.py from datetime import datetime from typing import Optional from sqlmodel import SQLModel, Field class Attachment(SQLModel, table=True): __tablename__ = "{prefix}_attachment" id: Optional[int] = Field(default=None, primary_key=True) filename: str = Field(max_length=255, description="原始文件名") file_path: str = Field(max_length=500, description="存储路径") file_url: str = Field(max_length=500, description="访问 URL") file_size: int = Field(description="文件大小（字节）") content_type: Optional[str] = Field(max_length=100, description="内容类型") file_ext: Optional[str] = Field(max_length=20, description="文件扩展名") category: str = Field(default="default", max_length=50, description="分类") storage_type: str = Field(max_length=20, description="存储类型") uploader_id: Optional[int] = Field(description="上传者 ID") is_compressed: bool = Field(default=False, description="是否已压缩") created_at: datetime = Field(default_factory=datetime.now, description="创建时间") deleted_at: Optional[datetime] = Field(default=None, description="软删除时间") class ChunkInfo(SQLModel, table=True): __tablename__ = "{prefix}_chunk_info" id: Optional[int] = Field(default=None, primary_key=True) upload_id: str = Field(max_length=64, unique=True, description="上传任务 ID") filename: str = Field(max_length=255, description="文件名") file_size: int = Field(description="文件总大小") chunk_size: int = Field(description="分片大小") total_chunks: int = Field(description="总分片数") status: int = Field(default=0, description="状态 0-上传中 1-已完成 2-已取消") created_at: datetime = Field(default_factory=datetime.now, description="创建时间") updated_at: datetime = Field(default_factory=datetime.now, description="更新时间")
接口契约要点
| 方法 | 路径 | 说明 |
|---|---|---|
| POST | /api/storage/upload | 上传文件（自动判断） |
| POST | /api/storage/upload/chunk/init | 初始化分片上传 |
| POST | /api/storage/upload/chunk/{upload_id}/{index} | 上传分片 |
| POST | /api/storage/upload/chunk/{upload_id}/merge | 合并分片 |
| GET | /api/storage/upload/chunk/{upload_id}/chunks | 获取已上传分片 |
| GET | /api/storage/download/{id} | 下载文件 |
| GET | /api/storage/preview/{id} | 预览文件 |
| GET | /api/storage/attachment/list | 附件列表 |
| GET | /api/storage/attachment/{id} | 附件详情 |
| DELETE | /api/storage/attachment/{id} | 删除附件 |
强制交付物
| 文档 | 位置 | 说明 |
|---|---|---|
| 接口契约 | api-contract-storage.md | 全量接口 |
| 接入指南 | docs/storage-module-guide.md | 表结构、配置、集成步骤 |
红线
- 不重复生成 FastAPI 基础骨架。
- 表名统一
{prefix}_attachment、{prefix}_chunk_info。 - 所有删除为软删除（
deleted_at）。 - 分片上传必须支持断点续传。
- 压缩功能必须可配置开关。
- 所有注释、文档用中文。
- 与
springboot-storage-module-skill保持 API 字段完全一致。
【考拉搞AI】，带你全面进入 VibeCoding 的世界~
触发关键词
静态资源模块、Python 文件上传、FastAPI 文件上传、大文件上传、 文件压缩、storage module、资源管理、分片上传、断点续传
