FastAPI 文件上传与数据混合处理教程

挑战：文件上传与结构化数据共存

在构建 web api 时，一个常见的需求是允许用户上传文件，同时提供与该文件相关的其他结构化数据。例如，上传一张图片的同时，附带该图片的事务 id 和组织 id。fastapi 提供了强大的类型提示和依赖注入系统来处理这类请求，但初学者可能会在如何将文件（uploadfile）与 basemodel 定义的结构化数据结合时遇到困惑。

一个常见的误区是尝试将 UploadFile 直接作为 Pydantic BaseModel 的字段：

from pydantic import BaseModel
from fastapi import UploadFile, File

class Example(BaseModel):
    image: UploadFile = File() # 这会导致错误
    transaction_id: str = None
    organization_id: str = None

这种做法会导致运行时错误，因为 UploadFile 对象本身不能被 Pydantic 直接序列化或反序列化为 JSON 格式。UploadFile 是一个特殊类型，FastAPI 会从 multipart/form-data 请求体中解析它，而不是从 JSON 请求体中。

FastAPI 的解决方案：混合参数处理

FastAPI 巧妙地处理了 multipart/form-data 请求。当你的端点函数签名中同时包含 UploadFile 类型参数和其他基本类型（如 str, int, bool）或使用 Form() 依赖的参数时，FastAPI 会自动识别这是一个 multipart/form-data 请求，并正确地解析各个部分。

以下是实现文件上传和附加数据的正确且推荐的方式：

from fastapi import FastAPI, UploadFile, Form
from typing import Annotated
import os
import shutil

app = FastAPI()

@app.post("/upload_file_and_data/")
async def upload_file_and_data(
    file: UploadFile, # 文件参数
    transaction_id: Annotated[str, Form()], # 附加数据参数，来自表单字段
    organization_id: Annotated[str, Form()] # 附加数据参数，来自表单字段
):
    """
    处理文件上传和附加数据的API端点。
    文件将被保存到服务器的 'uploaded_files' 目录中，
    文件名由 organization_id 和 transaction_id 组合而成。
    """
    # 1. 创建文件保存目录（如果不存在）
    upload_dir = "uploaded_files"
    os.makedirs(upload_dir, exist_ok=True)

    # 2. 构造安全的文件名
    # 清理输入，防止路径遍历攻击或非法字符
    safe_transaction_id = "".join(c for c in transaction_id if c.isalnum() or c in ('-', '_'))
    safe_organization_id = "".join(c for c in organization_id if c.isalnum() or c in ('-', '_'))

    # 获取原始文件扩展名
    file_extension = os.path.splitext(file.filename)[1] if file.filename else ""

    # 组合文件名，确保唯一性和可读性
    file_name = f"{safe_organization_id}_{safe_transaction_id}{file_extension}"
    file_path = os.path.join(upload_dir, file_name)

    try:
        # 3. 保存上传的文件
        # 使用 'wb' 模式以二进制写入，并使用 shutil.copyfileobj 进行高效复制
        with open(file_path, "wb") as buffer:
            shutil.copyfileobj(file.file, buffer) # file.file 是底层的 SpooledTemporaryFile 对象

        return {
            "message": f"文件 '{file.filename}' 已成功上传并保存为 '{file_name}'",
            "transaction_id": transaction_id,
            "organization_id": organization_id,
            "saved_path": file_path
        }
    except Exception as e:
        # 4. 错误处理
        return {"error": f"文件上传失败: {str(e)}"}

# 运行此应用
# 使用命令: uvicorn your_module_name:app --reload
# 可以通过 Swagger UI (http://127.0.0.1:8000/docs) 测试此端点

如何测试此端点：

由于这是一个 multipart/form-data 请求，你无法直接使用简单的 JSON 工具进行测试。推荐使用以下方法：

1. FastAPI 的 Swagger UI: 访问 http://127.0.0.1:8000/docs，找到 /upload_file_and_data/ 端点，你可以直接在浏览器界面中上传文件并填写其他表单字段。

2. curl 命令:

   curl -X POST "http://127.0.0.1:8000/upload_file_and_data/" \
        -H "accept: application/json" \
        -H "Content-Type: multipart/form-data" \
        -F "file=@/path/to/your/image.jpg" \
        -F "transaction_id=TXN12345" \
        -F "organization_id=ORG67890"

   请将 /path/to/your/image.jpg 替换为实际的文件路径。

   

   杰易OA办公自动化系统6.0

   基于Intranet/Internet 的Web下的办公自动化系统，采用了当今最先进的PHP技术，是综合大量用户的需求,经过充分的用户论证的基础上开发出来的，独特的即时信息、短信、电子邮件系统、完善的工作流、数据库安全备份等功能使得信息在企业内部传递效率极大提高，信息传递过程中耗费降到最低。办公人员得以从繁杂的日常办公事务处理中解放出来，参与更多的富于思考性和创造性的工作。系统力求突出体系结构简明

   下载

3. Python requests 库:

   import requests
   
   url = "http://127.0.0.1:8000/upload_file_and_data/"
   
   files = {'file': ('my_image.jpg', open('/path/to/your/image.jpg', 'rb'), 'image/jpeg')}
   data = {'transaction_id': 'TXN12345', 'organization_id': 'ORG67890'}
   
   response = requests.post(url, files=files, data=data)
   print(response.json())

注意事项与最佳实践

1. 文件保存策略：

   • 同步 vs. 异步： shutil.copyfileobj 是同步操作。对于非常大的文件，或者在高并发场景下，同步的文件 I/O 可能会阻塞事件循环。FastAPI 提供了 UploadFile.read() 和 UploadFile.write() 的异步版本，或者可以考虑使用 aiofiles 库进行异步文件操作。
   • 分块读取： 对于超大文件，不应一次性将整个文件读入内存。UploadFile.file 是一个文件类对象，你可以分块读取它（例如，在一个 while True 循环中调用 file.read(chunk_size)）。shutil.copyfileobj 内部已经实现了分块读取。
   • 临时文件： UploadFile 内部使用了 SpooledTemporaryFile，这意味着小文件会存在内存中，大文件会写入磁盘的临时文件。这有助于减少内存消耗。

2. 文件名安全与路径遍历：

   • 不要直接使用 file.filename 作为保存路径或文件名的一部分。 用户可以上传名为 ../../../../etc/passwd 的文件，这可能导致路径遍历攻击。
   • 清理和验证输入： 在上述示例中，我们对 transaction_id 和 organization_id 进行了简单的字符过滤。对于文件名本身，应该生成一个唯一且安全的文件名（例如，使用 UUID），并仅附加原始文件的扩展名（也要验证扩展名是否合法）。

3. 为什么 BaseModel 不适用 UploadFile：

   • BaseModel 主要用于解析 JSON 或其他结构化数据体。
   • UploadFile 代表的是 multipart/form-data 请求中的文件部分，它不是一个简单的字符串或数字，而是一个文件对象，需要特殊处理。
   • FastAPI 能够智能地区分请求体中的 JSON 部分（对应 BaseModel）和 form-data 部分（对应 UploadFile 和 Form() 参数）。

4. Form() 依赖的使用：

   • 在上面的例子中，我们使用了 Annotated[str, Form()]。Form() 是一个依赖，它告诉 FastAPI 这个参数应该从 multipart/form-data 的表单字段中获取。
   • 即使没有显式使用 Form()，如果你的端点函数签名中同时有 UploadFile 和其他基本类型参数（如 str），FastAPI 也会默认将这些基本类型参数视为来自表单字段。然而，显式使用 Form() 可以增加代码的清晰度和可读性。

5. 错误处理：

   • 除了文件上传失败，还应考虑其他潜在错误，如文件类型不匹配、文件大小超出限制、存储空间不足等。
   • FastAPI 允许你设置文件大小限制（例如，在 FastAPI 实例创建时通过 max_request_size 或使用 Depends 进行验证）。

总结

FastAPI 通过其直观的类型提示和依赖注入系统，使得同时处理文件上传和附加数据变得非常简单。核心思想是让 FastAPI 自动解析 multipart/form-data 请求，将文件部分映射到 UploadFile 参数，将其他表单字段映射到普通类型参数或使用 Form() 标记的参数。遵循这些最佳实践，可以构建出健壮、安全且高效的文件上传 API。