在 Web 应用开发中,文件上传功能是用户交互的重要组成部分,无论是上传图片、文档,还是其他类型的文件,都需要一个可靠且高效的解决方案。FastAPI 作为高性能的 Python Web 框架,凭借其简洁的语法和强大的功能,为文件上传提供了优雅且高效的实现方式。本文将深入探讨 FastAPI 中文件上传的各种场景与实践,带你从基础入门到进阶应用,全面掌握这一核心功能

一、FastAPI 文件上传基础:单文件上传

在 FastAPI 中,处理文件上传的关键在于FileUploadFile这两个工具。UploadFile是对SpooledTemporaryFile的封装,它支持异步操作,并且在性能和资源管理上表现出色,非常适合处理 HTTP 请求中的文件数据

1. 简单单文件上传示例

from fastapi import FastAPI, File, UploadFile

app = FastAPI()
@app.post("/uploadfile/")

async def create_upload_file(file: UploadFile = File(...)):
    return {"filename": file.filename}

在上述代码中:

  • @app.post("/uploadfile/")定义了一个处理 POST 请求的路由,路径为/uploadfile/
  • file: UploadFile = File(...)表示该接口接收一个文件,File(...)强调这是一个必填参数。
  • 接口返回值仅包含上传文件的文件名file.filename,实际应用中可根据需求进行更复杂的操作。

2. 读取文件内容

获取文件名只是第一步,更多时候我们需要读取文件内容。FastAPI 的 UploadFile 支持异步读取,示例如下:

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    contents = await file.read()
    file_size = len(contents)
    return {
        "filename": file.filename,
        "file_size": file_size,
        "content_type": file.content_type
    }

这里通过await file.read()异步读取文件内容,计算文件大小,并获取文件的content_type(文件类型),以便对文件进行更细致的处理和验证

3. 保存上传文件

实际应用中,我们通常需要将上传的文件保存到服务器。以下是一个将文件保存到本地指定目录的示例:

import os
from fastapi import FastAPI, File, UploadFile

app = FastAPI()

UPLOAD_DIR = "uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    file_path = os.path.join(UPLOAD_DIR, file.filename)
    with open(file_path, "wb") as buffer:
        buffer.write(await file.read())
    return {"message": "File saved successfully", "file_path": file_path}

代码中先创建保存文件的目录,再将文件内容写入到指定路径,确保文件能够持久化存储在服务器上

二、进阶应用:多文件上传与大文件处理

1. 多文件上传

当需要同时上传多个文件时,只需将UploadFile定义为列表形式,示例如下:​

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadfiles/")
async def create_upload_files(files: list[UploadFile] = File(...)):
    file_infos = []
    for file in files:
        contents = await file.read()
        file_infos.append({
            "filename": file.filename,
            "file_size": len(contents)
        })
    return {"uploaded_files": file_infos}

通过接收list[UploadFile]类型的参数,接口可以处理多个文件,遍历文件列表并获取每个文件的信息,实现多文件上传的功能

2. 大文件处理

对于大文件上传,直接一次性读取文件内容可能会导致内存不足。FastAPI 结合 Python 的文件操作,可以采用分块读取的方式处理大文件。以下是一个简单的分块读取示例:

from fastapi import FastAPI, File, UploadFile

app = FastAPI()

@app.post("/uploadlargefile/")
async def create_upload_large_file(file: UploadFile = File(...)):
    CHUNK_SIZE = 1024 * 1024  # 1MB
    file_size = 0
    with open("large_file.tmp", "wb") as buffer:
        while True:
            chunk = await file.read(CHUNK_SIZE)
            if not chunk:
                break
            buffer.write(chunk)
            file_size += len(chunk)
    return {"message": "Large file uploaded successfully", "file_size": file_size}

此示例中,每次读取固定大小(1MB)的文件块写入临时文件,直到文件读取完毕,有效降低了内存占用,适合处理大文件上传场景

三、文件上传的验证与安全

1. 文件类型验证

为了确保上传的文件符合要求,我们可以对文件类型进行验证。一种简单的方式是通过文件扩展名或content_type进行判断:

from fastapi import FastAPI, File, UploadFile, HTTPException

app = FastAPI()

ALLOWED_EXTENSIONS = {".jpg", ".jpeg", ".png"}
ALLOWED_CONTENT_TYPES = {"image/jpeg", "image/png"}

def is_allowed_file(filename, content_type):
    return (
        any(filename.endswith(ext) for ext in ALLOWED_EXTENSIONS)
        and content_type in ALLOWED_CONTENT_TYPES
    )

@app.post("/uploadimage/")
async def create_upload_image(file: UploadFile = File(...)):
    if not is_allowed_file(file.filename, file.content_type):
        raise HTTPException(status_code=400, detail="Invalid file type")
    return {"message": "Image uploaded successfully", "filename": file.filename}

2. 安全注意事项

  • 文件重命名:为避免文件名冲突和安全隐患,建议对上传的文件进行重命名,例如使用 UUID 作为文件名。
  • 存储路径限制:严格限制文件的保存路径,防止用户通过文件名或路径进行恶意操作(如目录穿越攻击)。
  • 病毒扫描:在生产环境中,对上传文件进行病毒扫描是保障系统安全的重要措施,可以集成第三方病毒扫描服务实现。
最后修改:2025 年 06 月 03 日
如果觉得我的文章对你有用,请随意赞赏