FastAPI 文件上传与图片管理实战:文件校验、存储与访问权限
在 Web 应用中,文件上传是非常常见的功能。例如:
-
用户上传头像
-
管理员上传文章封面
-
编辑人员上传文章配图
-
用户上传 PDF、Word 等附件
-
后台管理文件列表、删除文件
FastAPI 提供了 UploadFile 和 File,可以方便地接收上传文件。
不过,一个真正可用的文件上传模块,并不只是接收文件并保存到服务器,还需要考虑:
-
文件类型校验,避免上传危险文件。
-
文件大小限制,避免占用过多磁盘空间。
-
随机文件名,避免文件名冲突和路径穿越。
-
文件存储,区分文件访问地址与服务器物理路径。
-
访问权限,避免任何用户都能删除或管理文件。
-
图片访问,支持前端正常显示上传的图片。
-
异常处理,避免上传失败后留下无效文件。
本文将使用 FastAPI 实现一个完整的文件上传与图片管理模块。
一、准备开发环境
1. 创建项目
创建一个独立的 FastAPI 项目:
mkdir fastapi-file-manager
cd fastapi-file-manager
创建虚拟环境:
python3 -m venv venv
激活虚拟环境:
Linux:
source venv/bin/activate
Windows:
venv\Scripts\activate
2. 安装依赖
pip install fastapi uvicorn python-multipart pillow
各依赖的作用:
| 依赖 | 作用 |
|---|---|
| fastapi | Web API 框架 |
| uvicorn | ASGI 服务器 |
| python-multipart | 解析 multipart 文件上传请求 |
| pillow | 图片格式验证与图片处理 |
其中,python-multipart 是 FastAPI 接收表单文件上传所必需的依赖。
3. 创建项目目录
fastapi-file-manager/
├── app/
│ ├── __init__.py
│ ├── main.py
│ ├── config.py
│ ├── security.py
│ ├── storage.py
│ └── routers/
│ ├── __init__.py
│ └── files.py
├── uploads/
│ └── images/
└── requirements.txt
目录说明:
-
main.py:应用入口。 -
config.py:文件上传配置。 -
security.py:用户身份与权限依赖。 -
storage.py:文件保存和删除。 -
routers/files.py:文件上传、列表、查看和删除接口。 -
uploads/:文件存储目录。
这里使用本地磁盘作为存储介质。后续可以扩展为对象存储,例如 S3 或兼容 S3 的服务。
二、配置文件上传参数
首先创建 app/config.py:
from pathlib import Path
# 项目根目录
BASE_DIR = Path(__file__).resolve().parent.parent
# 文件存储目录
UPLOAD_DIR = BASE_DIR / "uploads"
# 图片存储目录
IMAGE_DIR = UPLOAD_DIR / "images"
# 最大上传文件大小:5 MB
MAX_FILE_SIZE = 5 * 1024 * 1024
# 允许上传的图片格式
ALLOWED_IMAGE_FORMATS = {
"JPEG",
"PNG",
"WEBP",
}
# 允许的 MIME 类型
ALLOWED_MIME_TYPES = {
"image/jpeg",
"image/png",
"image/webp",
}
# 创建存储目录
IMAGE_DIR.mkdir(parents=True, exist_ok=True)
这里限制上传文件最大为 5 MB,并且只允许 JPEG、PNG、WebP 三种图片格式。
需要注意:文件扩展名、浏览器提交的 MIME 类型都可以被伪造,因此不能仅凭这两项判断文件是否安全。
后面还会使用 Pillow 检查图片真实格式。
三、实现文件存储模块
创建 app/storage.py:
import uuid
from pathlib import Path
from app.config import IMAGE_DIR
def generate_filename(extension: str) -> str:
"""
生成随机文件名
"""
return f"{uuid.uuid4().hex}{extension}"
def save_file(content: bytes, extension: str) -> str:
"""
保存文件到本地磁盘
"""
filename = generate_filename(extension)
file_path = IMAGE_DIR / filename
file_path.write_bytes(content)
return filename
def delete_file(filename: str) -> bool:
"""
删除文件
"""
file_path = IMAGE_DIR / filename
if not file_path.is_file():
return False
file_path.unlink()
return True
def get_file_path(filename: str) -> Path:
"""
获取文件物理路径
"""
return IMAGE_DIR / filename
这里使用 UUID 生成随机文件名。
例如:
原始文件名:风景图片.png
保存文件名:
f1c9a8d4e7b24c2a9f6d3e8a1b7c5d20.png
为什么不直接使用用户上传的文件名?
主要有三个原因:
-
不同用户可能上传相同名称的文件。
-
用户可以伪造文件名,造成路径安全问题。
-
原始文件名可能包含特殊字符,不适合直接作为服务器存储名称。
因此,服务器应当使用随机生成的文件名,原始文件名只作为数据库中的展示信息保存。
四、实现图片文件校验
这是文件上传模块中非常重要的一部分。
不能只通过文件扩展名判断文件类型。
例如,用户上传一个 PHP 文件,却将文件名修改为:
shell.png
如果服务器只检查扩展名,就可能错误地将其当作图片处理。
因此,需要进行多层校验:
-
检查文件扩展名。
-
检查浏览器提交的 MIME 类型。
-
检查文件大小。
-
使用 Pillow 验证真实图片格式。
创建 app/routers/files.py,首先编写校验函数:
from io import BytesIO
from pathlib import Path
from fastapi import HTTPException, UploadFile
from PIL import Image, UnidentifiedImageError
from app.config import (
MAX_FILE_SIZE,
ALLOWED_IMAGE_FORMATS,
ALLOWED_MIME_TYPES,
)
async def validate_image(file: UploadFile) -> tuple[bytes, str]:
"""
验证图片文件
"""
# 检查 MIME 类型
if file.content_type not in ALLOWED_MIME_TYPES:
raise HTTPException(
status_code=400,
detail="不支持的图片类型"
)
# 检查扩展名
extension = Path(file.filename or "").suffix.lower()
allowed_extensions = {
".jpg",
".jpeg",
".png",
".webp",
}
if extension not in allowed_extensions:
raise HTTPException(
status_code=400,
detail="不支持的文件扩展名"
)
# 读取文件内容
content = await file.read(MAX_FILE_SIZE + 1)
# 检查文件大小
if len(content) > MAX_FILE_SIZE:
raise HTTPException(
status_code=413,
detail="文件大小不能超过 5 MB"
)
if not content:
raise HTTPException(
status_code=400,
detail="不能上传空文件"
)
# 验证图片真实格式
try:
with Image.open(BytesIO(content)) as image:
image.verify()
image_format = image.format
except (
UnidentifiedImageError,
OSError,
ValueError,
):
raise HTTPException(
status_code=400,
detail="文件不是有效的图片"
)
if image_format not in ALLOWED_IMAGE_FORMATS:
raise HTTPException(
status_code=400,
detail="图片真实格式不受支持"
)
# 返回文件内容和规范化扩展名
extension_map = {
"JPEG": ".jpg",
"PNG": ".png",
"WEBP": ".webp",
}
return content, extension_map[image_format]
为什么要读取 MAX_FILE_SIZE + 1 字节?
如果直接使用:
content = await file.read()
服务器会一次性读取整个文件。
当用户上传非常大的文件时,可能消耗大量内存。
这里最多读取 5 MB + 1 字节:
content = await file.read(MAX_FILE_SIZE + 1)
如果读取结果超过 5 MB,就可以立即拒绝。
需要注意,FastAPI 的 UploadFile 通常由底层 multipart 解析器先接收并暂存上传内容。因此,这种读取方式可以限制应用层读取和后续处理的内存消耗,但不能代替反向代理或 multipart 解析层的请求体大小限制。
另外,图片校验并不等于彻底安全。生产环境还应考虑图片解码资源消耗、恶意文件扫描,以及是否需要重新编码图片。
五、实现图片上传接口
接下来创建上传接口。
在 app/routers/files.py 中继续添加:
import asyncio
from fastapi import APIRouter, File, UploadFile, HTTPException
from app.storage import save_file
router = APIRouter(
prefix="/api/files",
tags=["文件管理"]
)
@router.post("/upload")
async def upload_image(
file: UploadFile = File(...)
):
"""
上传图片
"""
try:
content, extension = await validate_image(file)
# 将磁盘写入放到线程池执行
filename = await asyncio.to_thread(
save_file,
content,
extension
)
return {
"code": 200,
"message": "上传成功",
"data": {
"filename": filename,
"url": f"/media/images/{filename}",
"size": len(content),
"content_type": file.content_type,
}
}
finally:
await file.close()
这里使用 asyncio.to_thread() 执行磁盘写入,避免同步文件操作直接阻塞异步事件循环。
上传成功后,接口返回:
{
"code": 200,
"message": "上传成功",
"data": {
"filename": "f1c9a8d4e7b24c2a9f6d3e8a1b7c5d20.png",
"url": "/media/images/f1c9a8d4e7b24c2a9f6d3e8a1b7c5d20.png",
"size": 102400,
"content_type": "image/png"
}
}
前端可以直接使用返回的 URL 显示图片。
六、实现图片访问接口
图片上传成功后,还需要让浏览器能够访问它。
FastAPI 可以通过 StaticFiles 提供静态文件服务。
在 app/main.py 中:
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from app.config import IMAGE_DIR
from app.routers.files import router as files_router
app = FastAPI(
title="FastAPI 文件管理系统",
version="1.0.0"
)
# 注册上传接口
app.include_router(files_router)
# 提供图片访问
app.mount(
"/media/images",
StaticFiles(directory=str(IMAGE_DIR)),
name="images"
)
@app.get("/")
async def index():
return {
"message": "FastAPI 文件管理系统运行正常"
}
启动项目:
uvicorn app.main:app --reload
浏览器访问:
http://127.0.0.1:8000/docs
上传图片后,可以通过:
http://127.0.0.1:8000/media/images/文件名.png
访问图片。
为什么上传接口和图片访问接口要分开?
上传接口负责:
-
身份验证
-
权限验证
-
文件校验
-
文件保存
图片访问接口负责:
-
返回图片内容
-
设置 HTTP 响应类型
-
提供静态资源访问
两者职责不同。
如果图片是公开资源,可以使用静态文件服务。
如果图片包含私人资料、用户证件等敏感内容,就不能直接通过公开静态目录访问,而应该使用带身份验证的文件读取接口。
七、实现用户身份认证与访问权限
文件管理不能只考虑上传,还必须考虑谁有权操作文件。
例如:
| 用户角色 | 上传图片 | 查看图片 | 删除图片 |
|---|---|---|---|
| 普通用户 | 允许 | 自己的图片 | 自己的图片 |
| 编辑人员 | 允许 | 有权限的图片 | 自己的图片 |
| 管理员 | 允许 | 全部图片 | 全部图片 |
这里使用一个简单的身份认证依赖进行演示。
创建 app/security.py:
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
security = HTTPBearer()
# 教学示例:模拟用户数据
USERS = {
"user-1": {
"id": "user-1",
"username": "alice",
"role": "user",
},
"admin-1": {
"id": "admin-1",
"username": "admin",
"role": "admin",
},
}
async def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security)
):
"""
获取当前登录用户
注意:这里仅用于演示。
生产环境应验证 JWT 签名、有效期,
并从数据库查询用户。
"""
token = credentials.credentials
# 为了演示,使用固定 Token
user = USERS.get(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="登录状态无效"
)
return user
def require_admin(
user=Depends(get_current_user)
):
"""
管理员权限验证
"""
if user["role"] != "admin":
raise HTTPException(
status_code=403,
detail="没有管理员权限"
)
return user
注意:这里的固定 Token 只是为了让本篇示例能够独立运行,并不是真正的 JWT 认证。正式项目应替换为经过签名验证的 JWT,并从数据库读取用户当前状态和角色。
八、记录文件信息
仅仅将文件保存到磁盘还不够。
如果后续需要展示文件列表、搜索文件、删除文件,就需要记录文件元数据。
生产环境通常使用 MySQL 或 PostgreSQL 保存文件信息。
建议设计如下数据表:
CREATE TABLE files (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
user_id BIGINT NOT NULL,
original_name VARCHAR(255) NOT NULL,
filename VARCHAR(100) NOT NULL UNIQUE,
extension VARCHAR(20) NOT NULL,
mime_type VARCHAR(100) NOT NULL,
file_size BIGINT NOT NULL,
storage_path VARCHAR(500) NOT NULL,
created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
INDEX idx_user_id (user_id),
INDEX idx_created_at (created_at)
);
字段说明:
| 字段 | 作用 |
|---|---|
| id | 文件记录 ID |
| user_id | 上传用户 ID |
| original_name | 原始文件名 |
| filename | 服务器随机文件名 |
| extension | 文件扩展名 |
| mime_type | 文件 MIME 类型 |
| file_size | 文件大小 |
| storage_path | 文件存储位置 |
| created_at | 上传时间 |
这里需要区分两个概念:
文件本体保存在磁盘或对象存储中。
文件信息保存在数据库中。
不要将大文件直接存入普通数据库字段,除非业务确实需要使用数据库 BLOB 存储。
九、实现文件列表接口
假设数据库文件记录已经封装为 Repository 层,这里先使用内存列表演示接口逻辑。
在 app/routers/files.py 中增加:
from fastapi import Depends
from app.security import get_current_user, require_admin
# 教学示例:模拟文件记录
file_records = []
@router.get("")
async def list_files(
user=Depends(get_current_user)
):
"""
获取当前用户的文件列表
"""
records = [
item
for item in file_records
if item["user_id"] == user["id"]
]
return {
"code": 200,
"data": records
}
@router.get("/admin/all")
async def list_all_files(
user=Depends(require_admin)
):
"""
管理员查看全部文件
"""
return {
"code": 200,
"data": file_records
}
普通用户只能看到自己的文件。
管理员可以查看全部文件。
实际使用时,应该将 file_records 替换为数据库查询,并且在 SQL 查询中直接添加用户范围条件。
例如:
SELECT *
FROM files
WHERE user_id = ?
ORDER BY created_at DESC;
这样可以避免先查询全部文件,再在 Python 中筛选。
十、实现文件删除接口
删除文件时,需要同时检查:
-
文件记录是否存在。
-
当前用户是否有权限删除。
-
文件是否确实位于允许的存储目录中。
-
数据库记录与物理文件是否一致。
在 app/routers/files.py 中添加:
from app.storage import delete_file
@router.delete("/{file_id}")
async def remove_file(
file_id: int,
user=Depends(get_current_user)
):
"""
删除文件
"""
# 查找文件记录
record = next(
(
item
for item in file_records
if item["id"] == file_id
),
None
)
if not record:
raise HTTPException(
status_code=404,
detail="文件不存在"
)
# 检查文件所有权
if (
record["user_id"] != user["id"]
and user["role"] != "admin"
):
raise HTTPException(
status_code=403,
detail="无权删除该文件"
)
# 删除物理文件
filename = record["filename"]
await asyncio.to_thread(
delete_file,
filename
)
# 删除文件记录
file_records.remove(record)
return {
"code": 200,
"message": "删除成功"
}
这段代码演示了权限判断,但还没有实现完整的数据库持久化。
正式项目中,建议采用以下处理方式:
-
先在数据库事务中确认记录和权限。
-
删除文件时处理磁盘异常。
-
根据业务需要采用软删除,避免数据库与磁盘状态不一致。
-
定期清理没有数据库记录的孤立文件。
-
对删除操作记录审计日志。
还要注意:上面的演示代码假设 file_id 是唯一的,并且文件记录已经正确写入 file_records。实际上传接口应当在保存文件后创建对应的数据库记录。
十一、实现上传文件记录
为了让前面的内存演示流程完整,上传接口还需要将文件信息写入 file_records。
可以将上传接口调整为:
from datetime import datetime, timezone
from app.security import get_current_user
@router.post("/upload")
async def upload_image(
file: UploadFile = File(...),
user=Depends(get_current_user)
):
try:
content, extension = await validate_image(file)
filename = await asyncio.to_thread(
save_file,
content,
extension
)
record = {
"id": len(file_records) + 1,
"user_id": user["id"],
"original_name": file.filename or "image",
"filename": filename,
"extension": extension,
"mime_type": file.content_type,
"file_size": len(content),
"url": f"/media/images/{filename}",
"created_at": datetime.now(
timezone.utc
).isoformat()
}
file_records.append(record)
return {
"code": 200,
"message": "上传成功",
"data": record
}
finally:
await file.close()
现在上传、列表、删除之间已经建立了基本关联。
不过,内存列表存在明显缺陷:
-
服务重启后数据丢失。
-
多个 Uvicorn Worker 之间数据不共享。
-
多个进程可能生成重复 ID。
-
不适合生产环境。
因此,正式项目必须使用数据库保存文件记录。
十二、前端如何上传图片?
以 Vue 3 为例,前端可以使用 FormData 上传文件。
1. 创建上传组件
<template>
<div class="upload-box">
<input
type="file"
accept="image/jpeg,image/png,image/webp"
@change="handleFileChange"
/>
<button
:disabled="!selectedFile || uploading"
@click="uploadFile"
>
{{ uploading ? '上传中...' : '上传图片' }}
</button>
<div v-if="imageUrl">
<p>上传成功:</p>
<img
:src="imageUrl"
alt="上传图片"
width="200"
/>
</div>
</div>
</template>
<script setup>
import { ref } from 'vue'
import axios from 'axios'
const selectedFile = ref(null)
const imageUrl = ref('')
const uploading = ref(false)
function handleFileChange(event) {
selectedFile.value = event.target.files?.[0] || null
}
async function uploadFile() {
if (!selectedFile.value) {
return
}
const formData = new FormData()
formData.append('file', selectedFile.value)
uploading.value = true
try {
const response = await axios.post(
'/api/files/upload',
formData,
{
headers: {
Authorization: 'Bearer user-1'
}
}
)
imageUrl.value = response.data.data.url
} catch (error) {
console.error('上传失败', error)
} finally {
uploading.value = false
}
}
</script>
这里使用的 Bearer user-1 只是为了匹配本篇的固定 Token 演示。
正式项目应从登录状态中获取真正的 JWT,并由 Axios 请求拦截器统一添加认证头。
2. 为什么使用 FormData?
文件上传通常使用:
const formData = new FormData()
formData.append('file', file)
浏览器会自动生成:
Content-Type: multipart/form-data; boundary=...
不要手动将 Content-Type 固定为 multipart/form-data,否则可能丢失浏览器自动生成的 boundary 参数。
十三、生产环境如何部署文件存储?
开发环境中,直接使用本地目录即可:
项目目录/
└── uploads/
└── images/
生产环境建议将上传目录放在项目代码目录之外。
例如:
/var/www/
├── fastapi-app/
│ └── app/
│
└── uploads/
└── images/
这样可以避免代码更新、重新部署时误删用户上传的文件。
1. 修改文件存储路径
from pathlib import Path
UPLOAD_DIR = Path("/var/www/uploads")
IMAGE_DIR = UPLOAD_DIR / "images"
IMAGE_DIR.mkdir(
parents=True,
exist_ok=True
)
2. 设置目录权限
假设 FastAPI 使用 www-data 用户运行:
sudo mkdir -p /var/www/uploads/images
sudo chown -R www-data:www-data /var/www/uploads
sudo chmod -R 750 /var/www/uploads
具体用户和权限需要根据实际运行服务的用户调整。
不要为了方便直接使用:
chmod -R 777 /var/www/uploads
这会让其他本地用户也可能拥有不必要的写入权限。
3. 使用 Nginx 提供静态图片服务
如果图片是公开资源,可以交给 Nginx 直接读取。
在 Nginx 网站配置中增加:
location /media/images/ {
alias /var/www/uploads/images/;
expires 30d;
add_header Cache-Control "public, max-age=2592000";
access_log off;
autoindex off;
limit_except GET HEAD {
deny all;
}
try_files $request_filename =404;
}
注意:
-
alias指向实际图片目录。 -
autoindex off禁止目录列表。 -
limit_except只允许 GET 和 HEAD。 -
图片缓存时间可以根据业务调整。
使用 Nginx 提供静态文件后,可以从 main.py 中移除对应的 StaticFiles 挂载,避免两套服务重复处理同一个路径。
重要: Nginx 静态文件服务适用于公开图片。如果图片需要登录才能访问,不能直接使用这个公开目录配置,而应通过带权限验证的接口或受控的内部文件服务读取。
十四、文件上传安全注意事项
文件上传接口是 Web 应用中比较容易被滥用的入口。
生产环境至少需要关注以下问题。
1. 不要信任用户文件名
不要直接使用:
file_path = UPLOAD_DIR / file.filename
应该使用服务器生成的随机文件名。
2. 不要只检查扩展名
以下检查是不够的:
if file.filename.endswith(".png"):
pass
扩展名可以伪造。
应该结合 MIME 类型、文件内容和真实格式检查。
3. 限制文件大小
除了 FastAPI 应用层,还应在 Nginx 限制请求体:
client_max_body_size 6m;
如果应用允许单张图片最大 5 MB,可以将 Nginx 限制稍微放宽,为 multipart 表单额外开销留出空间。
4. 上传目录不要执行脚本
上传文件应该放在 Web 程序代码目录之外。
不要让用户上传的文件被 PHP、Python、CGI 等解释器执行。
如果使用 Nginx 提供文件服务,还应避免将上传目录配置为可执行脚本的目录。
5. 防止路径穿越
不要将用户传入的文件名直接拼接到服务器路径中。
对于需要根据文件名读取文件的接口,应该只接受数据库中已登记的文件标识,并由服务端取得实际存储路径。
6. 考虑恶意图片和资源消耗
真实图片格式校验并不能完全防止恶意文件。
生产环境还可以:
-
限制图片最大宽度和高度。
-
限制像素总量,防止解压缩炸弹。
-
使用 Pillow 重新解码并编码图片。
-
对文件进行病毒扫描。
-
限制单个用户的上传频率。
-
对上传接口增加并发和请求速率限制。
7. 私有图片需要权限验证
不要把身份证、合同、个人资料等文件放在公开静态目录中。
私有文件应通过用户身份验证和资源所有权验证后再返回文件内容。
十五、常见问题
问题一:上传文件时报错 422
常见原因:
-
没有安装
python-multipart。 -
前端没有使用
FormData。 -
前端字段名称与后端参数名称不一致。
检查依赖:
pip install python-multipart
前后端字段名称必须一致:
file: UploadFile = File(...)
formData.append('file', file)
问题二:上传后图片无法显示
检查:
-
文件是否真的保存成功。
-
返回的图片 URL 是否正确。
-
Nginx 是否配置了对应的图片目录。
-
文件权限是否允许 Nginx 读取。
-
是否存在反向代理路径冲突。
可以直接使用 curl 测试:
curl -I https://example.com/media/images/test.png
问题三:上传大文件时出现 413
413 表示请求实体过大。
检查 Nginx:
client_max_body_size 6m;
同时检查 FastAPI 中的文件大小限制。
两层限制都需要合理配置。
问题四:服务器重启后文件列表为空
如果使用的是本文的内存列表:
file_records = []
重启后数据自然会丢失。
应当将文件元数据写入 MySQL 或 PostgreSQL,不能依赖 Python 全局变量保存业务数据。
问题五:删除文件后图片仍然可以访问
检查:
-
是否删除了正确的物理文件。
-
Nginx 是否配置了较长的浏览器缓存。
-
CDN 是否缓存了旧图片。
-
是否存在多个存储节点或副本。
必要时执行 CDN 缓存清理。
十六、总结
本文从零实现了 FastAPI 文件上传与图片管理的基本流程。
主要完成了:
-
使用
UploadFile接收文件。 -
使用
python-multipart解析上传请求。 -
限制文件大小和允许的图片格式。
-
使用 Pillow 验证真实图片格式。
-
使用 UUID 生成随机文件名。
-
将文件保存到服务器本地磁盘。
-
使用 FastAPI 静态文件服务访问图片。
-
实现用户身份验证和资源所有权检查。
-
设计文件元数据数据库表。
-
实现文件列表和删除接口。
-
使用 Vue 3 的 FormData 上传图片。
-
使用 Nginx 提供静态图片服务。
需要明确的是,本文的内存用户、文件记录和固定 Token 都是为了展示接口流程的教学实现,并非完整的生产级系统。正式上线前,应将用户认证替换为真实 JWT,将文件记录持久化到数据库,并完善上传限流、权限检查、磁盘异常处理和文件扫描。
- 点赞
- 收藏
- 关注作者
评论(0)