from fastapi import FastAPI, UploadFile, File, HTTPException
from fastapi.responses import FileResponse
import shutil
from pathlib import Path
import uuid
import os
app = FastAPI()
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
@app.post("/upload/")
async def upload_file(file: UploadFile = File(...)):
"""基础文件上传"""
# 验证文件类型
allowed_types = ["image/jpeg", "image/png", "image/gif", "application/pdf"]
if file.content_type not in allowed_types:
raise HTTPException(status_code=400, detail="不支持的文件类型")
# 生成安全文件名
file_extension = file.filename.split(".")[-1]
safe_filename = f"{uuid.uuid4().hex}.{file_extension}"
# 保存文件
file_path = UPLOAD_DIR / safe_filename
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {
"filename": file.filename,
"saved_as": safe_filename,
"size": file_path.stat().st_size,
"content_type": file.content_type
}

from typing import List
@app.post("/upload/multiple/")
async def upload_multiple_files(
files: List[UploadFile] = File(...)
):
"""多文件上传"""
results = []
for file in files:
# 验证
if file.content_type not in ["image/jpeg", "image/png"]:
continue
# 保存
file_extension = file.filename.split(".")[-1]
safe_filename = f"{uuid.uuid4().hex}.{file_extension}"
file_path = UPLOAD_DIR / safe_filename
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
results.append({
"original": file.filename,
"saved_as": safe_filename,
"size": file_path.stat().st_size
})
return {"uploaded": results, "count": len(results)}

from fastapi import Form
@app.post("/upload/profile/")
async def upload_profile(
file: UploadFile = File(...),
user_id: int = Form(...),
description: str = Form(...)
):
"""带表单数据的文件上传"""
# 验证用户
if user_id <= 0:
raise HTTPException(status_code=400, detail="无效的用户ID")
# 验证文件
if not file.filename.endswith((".jpg", ".jpeg", ".png")):
raise HTTPException(status_code=400, detail="只支持图片")
# 保存
safe_filename = f"{uuid.uuid4().hex}.jpg"
file_path = UPLOAD_DIR / "profiles" / safe_filename
file_path.parent.mkdir(exist_ok=True)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {
"user_id": user_id,
"description": description,
"avatar": safe_filename
}

import magic  # pip install python-magic
ALLOWED_MIME_TYPES = {
"image/jpeg": [".jpg", ".jpeg"],
"image/png": [".png"],
"image/gif": [".gif"],
"application/pdf": [".pdf"],
"text/plain": [".txt"],
}
def validate_file_type(file: UploadFile) -> tuple[bool, str]:
"""验证文件类型"""
# 检查扩展名
if not file.filename:
return False, "文件名为空"
extension = "." + file.filename.split(".")[-1].lower()
# 检查MIME类型
if file.content_type not in ALLOWED_MIME_TYPES:
return False, f"不支持的文件类型: {file.content_type}"
# 检查扩展名是否匹配MIME
if extension not in ALLOWED_MIME_TYPES.get(file.content_type, []):
return False, "文件扩展名与类型不匹配"
return True, "验证通过"
使用
@app.post("/upload/validated/")
async def upload_validated(file: UploadFile = File(...)):
is_valid, message = validate_file_type(file)
if not is_valid:
raise HTTPException(status_code=400, detail=message)
# 继续处理...

MAX_FILE_SIZE = 10  1024  1024  # 10MB
def validate_file_size(file: UploadFile, max_size: int = MAX_FILE_SIZE) -> tuple[bool, str]:
"""验证文件大小"""
# 读取文件内容获取大小
content = file.file.read()
file.file.seek(0)  # 重置指针
if len(content) > max_size:
return False, f"文件过大，最大允许 {max_size // 1024 // 1024}MB"
if len(content) == 0:
return False, "文件为空"
return True, "验证通过"
使用
@app.post("/upload/size-checked/")
async def upload_size_checked(file: UploadFile = File(...)):
is_valid, message = validate_file_size(file)
if not is_valid:
raise HTTPException(status_code=400, detail=message)
# 继续处理...

import re
def sanitize_filename(filename: str) -> str:
"""清理文件名"""
if not filename:
raise ValueError("文件名为空")
# 移除路径遍历字符
filename = os.path.basename(filename)
# 移除特殊字符
filename = re.sub(r'[<>:"/\\|?*]', '_', filename)
# 限制长度
if len(filename) > 255:
name, ext = os.path.splitext(filename)
filename = name[:255-len(ext)] + ext
return filename
def generate_safe_filename(original: str) -> str:
"""生成安全的文件名"""
# 获取扩展名
extension = os.path.splitext(original)[1].lower()
# 生成UUID
safe_name = f"{uuid.uuid4().hex}{extension}"
return safe_name

from dataclasses import dataclass
from typing import Optional
@dataclass
class FileValidationResult:
is_valid: bool
error: Optional[str] = None
safe_filename: Optional[str] = None
file_size: Optional[int] = None
class FileValidator:
"""文件验证器"""
def __init__(
self,
allowed_types: list[str] = None,
max_size: int = 10  1024  1024,
allowed_extensions: list[str] = None
):
self.allowed_types = allowed_types or ["image/jpeg", "image/png", "image/gif"]
self.max_size = max_size
self.allowed_extensions = allowed_extensions or [".jpg", ".jpeg", ".png", ".gif"]
def validate(self, file: UploadFile) -> FileValidationResult:
# 检查文件名
if not file.filename:
return FileValidationResult(is_valid=False, error="文件名为空")
# 检查扩展名
extension = os.path.splitext(file.filename)[1].lower()
if extension not in self.allowed_extensions:
return FileValidationResult(
is_valid=False,
error=f"不支持的文件扩展名: {extension}"
)
# 检查MIME类型
if file.content_type not in self.allowed_types:
return FileValidationResult(
is_valid=False,
error=f"不支持的文件类型: {file.content_type}"
)
# 检查大小
content = file.file.read()
file.file.seek(0)
if len(content) > self.max_size:
return FileValidationResult(
is_valid=False,
error=f"文件过大，最大允许 {self.max_size // 1024 // 1024}MB"
)
if len(content) == 0:
return FileValidationResult(
is_valid=False,
error="文件为空"
)
# 生成安全文件名
safe_filename = f"{uuid.uuid4().hex}{extension}"
return FileValidationResult(
is_valid=True,
safe_filename=safe_filename,
file_size=len(content)
)
使用
validator = FileValidator(
allowed_types=["image/jpeg", "image/png"],
max_size=5  1024  1024,  # 5MB
allowed_extensions=[".jpg", ".jpeg", ".png"]
)
@app.post("/upload/strict/")
async def upload_strict(file: UploadFile = File(...)):
result = validator.validate(file)
if not result.is_valid:
raise HTTPException(status_code=400, detail=result.error)
# 保存文件
file_path = UPLOAD_DIR / result.safe_filename
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
return {
"filename": result.safe_filename,
"size": result.file_size
}

from fastapi.responses import FileResponse, StreamingResponse
from pathlib import Path
@app.get("/download/{filename}")
async def download_file(filename: str):
"""下载文件"""
# 验证文件名
safe_filename = os.path.basename(filename)
file_path = UPLOAD_DIR / safe_filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
return FileResponse(
path=file_path,
filename=safe_filename,
media_type="application/octet-stream"
)

def file_iterator(file_path: Path, chunk_size: int = 8192):
"""文件迭代器"""
with open(file_path, "rb") as f:
while chunk := f.read(chunk_size):
yield chunk
@app.get("/download/large/{filename}")
async def download_large_file(filename: str):
"""大文件流式下载"""
safe_filename = os.path.basename(filename)
file_path = UPLOAD_DIR / safe_filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
return StreamingResponse(
file_iterator(file_path),
media_type="application/octet-stream",
headers={
"Content-Disposition": f'attachment; filename="{safe_filename}"'
}
)

from fastapi import Depends
def get_current_user(token: str = Depends(oauth2_scheme)):
# 验证token，返回用户信息
...
@app.get("/files/{file_id}/download")
async def download_file(
file_id: int,
current_user: dict = Depends(get_current_user)
):
"""带权限的文件下载"""
# 从数据库获取文件信息
file_info = get_file_from_db(file_id)
if not file_info:
raise HTTPException(status_code=404, detail="文件不存在")
# 检查权限
if file_info.owner_id != current_user["id"] and not current_user.get("is_admin"):
raise HTTPException(status_code=403, detail="无权访问")
file_path = Path(file_info.file_path)
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
return FileResponse(
path=file_path,
filename=file_info.original_name,
media_type=file_info.content_type
)

from PIL import Image
import io
@app.post("/upload/image/")
async def upload_image(file: UploadFile = File(...)):
"""上传图片并生成缩略图"""
# 验证
if file.content_type not in ["image/jpeg", "image/png", "image/gif"]:
raise HTTPException(status_code=400, detail="只支持图片")
# 读取图片
image = Image.open(file.file)
# 验证图片有效性
image.verify()
file.file.seek(0)
image = Image.open(file.file)
# 生成缩略图
thumbnail = image.copy()
thumbnail.thumbnail((200, 200))
# 保存原图
original_filename = f"{uuid.uuid4().hex}.jpg"
original_path = UPLOAD_DIR / "original" / original_filename
original_path.parent.mkdir(exist_ok=True)
image.save(original_path, quality=95)
# 保存缩略图
thumbnail_filename = f"{uuid.uuid4().hex}_thumb.jpg"
thumbnail_path = UPLOAD_DIR / "thumbnail" / thumbnail_filename
thumbnail_path.parent.mkdir(exist_ok=True)
thumbnail.save(thumbnail_path, quality=80)
return {
"original": original_filename,
"thumbnail": thumbnail_filename,
"size": image.size
}

@app.post("/convert/")
async def convert_image(
file: UploadFile = File(...),
format: str = "webp"
):
"""转换图片格式"""
if file.content_type not in ["image/jpeg", "image/png", "image/gif"]:
raise HTTPException(status_code=400, detail="只支持图片")
# 读取图片
image = Image.open(file.file)
# 转换格式
output = io.BytesIO()
format_map = {
"jpg": "JPEG",
"jpeg": "JPEG",
"png": "PNG",
"webp": "WEBP",
"gif": "GIF"
}
image.save(output, format=format_map.get(format.lower(), "JPEG"))
output.seek(0)
return StreamingResponse(
output,
media_type=f"image/{format}",
headers={"Content-Disposition": f'attachment; filename="converted.{format}"'}
)

import boto3
from botocore.exceptions import ClientError
s3_client = boto3.client(
"s3",
aws_access_key_id=os.getenv("AWS_ACCESS_KEY"),
aws_secret_access_key=os.getenv("AWS_SECRET_KEY"),
region_name="us-east-1"
)
BUCKET_NAME = "my-app-uploads"
@app.post("/upload/s3/")
async def upload_to_s3(file: UploadFile = File(...)):
"""上传到S3"""
# 验证
result = validator.validate(file)
if not result.is_valid:
raise HTTPException(status_code=400, detail=result.error)
# 生成S3对象键
s3_key = f"uploads/{result.safe_filename}"
try:
s3_client.upload_fileobj(
file.file,
BUCKET_NAME,
s3_key,
ExtraArgs={
"ContentType": file.content_type,
"ACL": "private"
}
)
return {
"key": s3_key,
"url": f"https://{BUCKET_NAME}.s3.amazonaws.com/{s3_key}"
}
except ClientError as e:
raise HTTPException(status_code=500, detail=str(e))

@app.get("/files/{s3_key}/url")
async def get_presigned_url(s3_key: str):
"""获取预签名URL"""
try:
url = s3_client.generate_presigned_url(
"get_object",
Params={
"Bucket": BUCKET_NAME,
"Key": s3_key
},
ExpiresIn=3600  # 1小时有效
)
return {"url": url}
except ClientError as e:
raise HTTPException(status_code=500, detail=str(e))

class SecureFileHandler:
"""安全文件处理器"""
def __init__(self, upload_dir: Path):
self.upload_dir = upload_dir.resolve()  # 获取绝对路径
def save_file(self, file: UploadFile) -> Path:
"""安全保存文件"""
# 1. 验证文件
result = validator.validate(file)
if not result.is_valid:
raise ValueError(result.error)
# 2. 生成安全文件名
safe_filename = result.safe_filename
# 3. 确保路径在上传目录内
file_path = (self.upload_dir / safe_filename).resolve()
# 检查路径遍历攻击
if not str(file_path).startswith(str(self.upload_dir)):
raise ValueError("非法的文件路径")
# 4. 确保目录存在
file_path.parent.mkdir(parents=True, exist_ok=True)
# 5. 保存文件
with open(file_path, "wb") as f:
shutil.copyfileobj(file.file, f)
# 6. 设置文件权限
os.chmod(file_path, 0o644)
return file_path
def serve_file(self, filename: str) -> Path:
"""安全提供文件"""
# 1. 清理文件名
safe_filename = os.path.basename(filename)
# 2. 构建路径
file_path = (self.upload_dir / safe_filename).resolve()
# 3. 检查路径遍历
if not str(file_path).startswith(str(self.upload_dir)):
raise ValueError("非法的文件路径")
# 4. 检查存在
if not file_path.exists():
raise FileNotFoundError("文件不存在")
return file_path

from fastapi import Request, HTTPException
import secrets
生成上传令牌
def generate_upload_token() -> str:
return secrets.token_urlsafe(32)
存储令牌（生产环境用Redis）
upload_tokens: dict = {}
@app.post("/upload/protected/")
async def upload_protected(
file: UploadFile = File(...),
token: str = Form(...),
request: Request = None
):
"""带CSRF保护的文件上传"""
# 验证令牌
if token not in upload_tokens:
raise HTTPException(status_code=403, detail="无效的令牌")
# 验证令牌来源
stored_token = upload_tokens[token]
if stored_token["session_id"] != request.session.get("session_id"):
raise HTTPException(status_code=403, detail="令牌不匹配")
# 验证文件
result = validator.validate(file)
if not result.is_valid:
raise HTTPException(status_code=400, detail=result.error)
# 删除已使用的令牌
del upload_tokens[token]
# 保存文件...

from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, status
from fastapi.responses import FileResponse
from pydantic import BaseModel
from typing import Optional
from datetime import datetime
from pathlib import Path
import uuid
import os
app = FastAPI()
配置
UPLOAD_DIR = Path("uploads")
MAX_FILE_SIZE = 50  1024  1024  # 50MB
ALLOWED_TYPES = ["image/jpeg", "image/png", "image/gif", "application/pdf"]
数据库模型（简化）
class FileRecord(BaseModel):
id: str
original_name: str
saved_name: str
size: int
content_type: str
owner_id: int
created_at: datetime
is_public: bool = False
内存存储（生产环境用数据库）
files_db: dict = {}
class FileValidator:
"""文件验证器"""
def validate(self, file: UploadFile) -> tuple[bool, str, str]:
# 检查扩展名
allowed_extensions = [".jpg", ".jpeg", ".png", ".gif", ".pdf"]
extension = os.path.splitext(file.filename)[1].lower()
if extension not in allowed_extensions:
return False, f"不支持的文件类型: {extension}", ""
# 检查MIME
if file.content_type not in ALLOWED_TYPES:
return False, f"不支持的文件类型: {file.content_type}", ""
# 检查大小
content = file.file.read()
file.file.seek(0)
if len(content) > MAX_FILE_SIZE:
return False, f"文件过大，最大允许 {MAX_FILE_SIZE // 1024 // 1024}MB", ""
if len(content) == 0:
return False, "文件为空", ""
# 生成安全文件名
safe_name = f"{uuid.uuid4().hex}{extension}"
return True, "验证通过", safe_name
validator = FileValidator()
class FileUploadResponse(BaseModel):
id: str
original_name: str
saved_name: str
size: int
content_type: str
url: str
@app.post("/files/upload/", response_model=FileUploadResponse)
async def upload_file(
file: UploadFile = File(...),
owner_id: int = Depends(lambda: 1)  # 从认证获取
):
"""上传文件"""
# 验证
is_valid, message, safe_name = validator.validate(file)
if not is_valid:
raise HTTPException(status_code=400, detail=message)
# 保存
file_path = UPLOAD_DIR / safe_name
file_path.parent.mkdir(exist_ok=True)
with open(file_path, "wb") as buffer:
shutil.copyfileobj(file.file, buffer)
# 创建记录
file_id = str(uuid.uuid4())
record = FileRecord(
id=file_id,
original_name=file.filename,
saved_name=safe_name,
size=file_path.stat().st_size,
content_type=file.content_type,
owner_id=owner_id,
created_at=datetime.utcnow()
)
files_db[file_id] = record
return FileUploadResponse(
id=file_id,
original_name=file.filename,
saved_name=safe_name,
size=record.size,
content_type=record.content_type,
url=f"/files/{file_id}/download"
)
@app.get("/files/{file_id}/download")
async def download_file(file_id: str):
"""下载文件"""
record = files_db.get(file_id)
if not record:
raise HTTPException(status_code=404, detail="文件不存在")
# 检查权限
# if record.owner_id != current_user_id and not record.is_public:
#     raise HTTPException(status_code=403, detail="无权访问")
file_path = UPLOAD_DIR / record.saved_name
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
return FileResponse(
path=file_path,
filename=record.original_name,
media_type=record.content_type
)
@app.get("/files/{file_id}")
async def get_file_info(file_id: str):
"""获取文件信息"""
record = files_db.get(file_id)
if not record:
raise HTTPException(status_code=404, detail="文件不存在")
return record
@app.delete("/files/{file_id}")
async def delete_file(file_id: str, owner_id: int = Depends(lambda: 1)):
"""删除文件"""
record = files_db.get(file_id)
if not record:
raise HTTPException(status_code=404, detail="文件不存在")
# 检查权限
if record.owner_id != owner_id:
raise HTTPException(status_code=403, detail="无权删除")
# 删除文件
file_path = UPLOAD_DIR / record.saved_name
if file_path.exists():
file_path.unlink()
# 删除记录
del files_db[file_id]
return {"message": "文件已删除"}
@app.get("/files/")
async def list_files(
owner_id: int = Depends(lambda: 1),
limit: int = 20,
skip: int = 0
):
"""列出文件"""
user_files = [
f for f in files_db.values()
if f.owner_id == owner_id
][skip:skip + limit]
return {
"files": user_files,
"total": len([f for f in files_db.values() if f.owner_id == owner_id])
}

📚 推荐阅读

KEYWORDS

IL, Python, python, 类, 对象

💡 如果你觉得这篇文章有帮助，请点个在看，分享给更多需要的人！

📝 关注我，获取更多实用干货～

🤝 有问题欢迎评论区留言交流！