网站导航
> 技术文档 > Python----FastAPI（表单数据、异步处理、文件上传、请求对象Request）_python fastapi request

Python----FastAPI（表单数据、异步处理、文件上传、请求对象Request）_python fastapi request

网友: 技术文档 2025-09-03 12:36:20

一、表单数据

要使用表单，需预先安装python-multipart

pip install python-multipart==0.0.20

提示

自 FastAPI 版本 0.113.0 起支持此功能

# 导入 FastAPI 核心类from fastapi import FastAPI, Form# 导入 Pydantic 模型类，用于数据校验和类型提示from pydantic import BaseModel# 导入 typing 模块的 Annotated，用于添加额外的元数据（如 Form()）from typing import Annotated# 创建一个 FastAPI 应用实例app = FastAPI()# ---# 第一个登录接口：使用 Form() 直接在函数参数中声明表单字段@app.post(\'/login1\')def login(username: str = Form(...), password: str = Form(...)): \"\"\" 处理 /login1 路由的 POST 请求。 参数说明： - username: 使用 Form(...) 声明，表示该参数是一个必填的表单字段。 - password: 同样是必填的表单字段。 功能： - 直接从请求体（form-data）中获取 username 和 password。 - 返回一个包含这些数据的字典。 \"\"\" return {\'username\': username, \'password\': password}# ---# 第二个登录接口：使用 Pydantic 模型和 Annotated[..., Form()]# 定义一个 Pydantic 模型，用于规范请求体中的数据格式class User1(BaseModel): username: str password: str@app.post(\'/login2\')def login2(user: Annotated[User1, Form()]): \"\"\" 处理 /login2 路由的 POST 请求。 思路： - 定义 User1 模型来规定数据结构。 - 在函数参数中，使用 Annotated[User1, Form()] 告诉 FastAPI： - 该参数 `user` 应该是一个 User1 类型的对象。 - 这个对象中的字段（username, password）应该从请求的表单数据（form-data）中提取。 功能： - FastAPI 会自动解析表单数据，并将其转换为一个 User1 实例。 - 直接返回这个完整的 User1 对象。 \"\"\" # FastAPI 自动将表单数据转换为 User1 对象 return user# ---# 第三个登录接口：将 Form(...) 直接放在 Pydantic 模型中# 定义一个 Pydantic 模型，并在其字段中声明 Form()class User2(BaseModel): username: str = Form(...) # 在模型字段中直接声明它是表单数据 password: str = Form(...) # 同样是表单数据@app.post(\'/login3\')def login3(user: Annotated[User2, Form()]): \"\"\" 处理 /login3 路由的 POST 请求。 思路： - 定义 User2 模型，并在字段上直接使用 `Form(...)`，这种方式更加简洁。 - 在函数参数中，同样使用 Annotated[User2, Form()]。 - 这表示 FastAPI 需要从表单数据中创建一个 User2 实例。 - 注意：当模型字段本身已经声明了 `Form()` 时，函数参数中的 `Annotated[..., Form()]` 实际上不是必需的，但保留它会使代码意图更清晰。 功能： - 与 /login2 功能相同，自动将表单数据解析并转换为 User2 实例。 - 返回这个完整的 User2 对象。 \"\"\" return user# ---# 运行服务器if __name__ == \'__main__\': # 导入 uvicorn 库，这是一个 ASGI 服务器 import uvicorn # 运行 FastAPI 应用 # app=\'demo10:app\' 表示在 demo10.py 文件中寻找名为 app 的应用实例 # host=\'127.0.0.1\' 表示监听本地回环地址 # port=8000 表示监听 8000 端口 # reload=True 表示文件修改后自动重启服务器，方便开发 uvicorn.run(app=\'demo10:app\', host=\'127.0.0.1\', port=8000, reload=True)

二、异步处理

        在 FastAPI 中，异步（async）和非异步（同步）编程方式是其核心特性之一，它们在处理请求、性能以及并发能力上有着显著的区别。

异步 (async def)

  • 事件循环：异步代码运行在 Python 的异步事件循环（如 asyncio）中。事件循环负责协调多个协程（coroutines），在某个协程等待 I/O 操作（如数据库查询或 HTTP 请求）时，事件循环可以切换到其他协程执行。
  • 非阻塞：当一个异步函数调用 await，它会暂停执行，将控制权交回事件循环，允许其他任务运行，直到等待的操作完成。
  • 并发性：异步模式允许单个线程处理大量并发请求，特别适合高并发场景（如 Web 服务器处理大量客户端请求）。

非异步 (def)

  • 阻塞式执行：同步函数在调用时会完全占用线程，直到函数执行完成才会释放线程。
  • 线程池：在 FastAPI 中，同步函数由工作线程（worker threads）处理，Uvicorn（FastAPI 常用的 ASGI 服务器）会将同步函数放入线程池运行。
  • 并发限制：线程池的大小限制了同步函数的并发能力。如果线程池耗尽（例如，处理大量阻塞请求），新请求将排队等待。
# 导入 FastAPI 核心类from fastapi import FastAPI# 导入 asyncio 模块，用于处理并发和异步任务import asyncio# 导入 time 模块，用于测量代码执行时间import time# 创建一个 FastAPI 应用实例app = FastAPI()# ---# 异步 endpoint：模拟并发 I/O 操作@app.get(\"/async\")async def async_endpoint(): \"\"\" 处理 /async 路由的 GET 请求。 代码思路： - 使用 `async def` 关键字将此函数定义为异步函数。 - 记录开始时间。 - 创建一个包含 5 个异步任务的列表，每个任务使用 `asyncio.sleep(1)` 模拟一个耗时 1 秒的 I/O 操作。 - 使用 `await asyncio.gather(*tasks)` 并发地（concurrently）运行所有任务。这意味着所有任务会几乎同时开始，程序会等待所有任务都完成后再继续。 - 记录结束时间，并计算总耗时。 实现功能： - 演示了 FastAPI 如何高效地处理异步任务。 - 尽管有 5 个耗时 1 秒的操作，但由于它们是并发执行的，整个函数的执行时间约等于单个最长任务的时间，即 **约 1 秒**。 - 这对于处理网络请求、数据库查询等 I/O 密集型任务非常有用，因为它不会阻塞整个服务器。 \"\"\" start = time.time() # 模拟 5 次异步 I/O 操作（并发执行） tasks = [asyncio.sleep(1) for _ in range(5)] # 使用 gather 并发执行所有任务 await asyncio.gather(*tasks) end = time.time() return {\"异步时长\": f\"{end - start:.2f}秒\"}# ---# 同步 endpoint：模拟相同的 I/O 操作@app.get(\"/sync\")def sync_endpoint(): \"\"\" 处理 /sync 路由的 GET 请求。 代码思路： - 这是一个普通的同步函数。 - 记录开始时间。 - 使用一个 `for` 循环，通过 `time.sleep(1)` 模拟 5 次耗时 1 秒的操作。 - `time.sleep()` 是一个同步阻塞函数，它会暂停整个程序的执行，直到指定的秒数过去。 - 因此，每次循环都会等待 1 秒，然后才执行下一次循环，是顺序执行的。 - 记录结束时间，并计算总耗时。 实现功能： - 演示了同步操作会阻塞执行。 - 尽管操作内容与异步函数相同，但由于是顺序执行的，总耗时约等于所有单个任务时间的总和，即 **约 5 秒**。 - 当一个同步端点被请求时，它会占用一个工作线程，直到任务完成。如果任务耗时很长，它会阻塞该线程，导致无法处理其他请求。 \"\"\" start = time.time() # 模拟 5 次同步 I/O 操作（顺序执行） for _ in range(5): time.sleep(1) end = time.time() return {\"同步时长\": f\"{end - start:.2f}秒\"}# ---# 运行服务器if __name__ == \"__main__\": # 导入 uvicorn，这是一个 ASGI 服务器，用于运行 FastAPI 应用 import uvicorn # uvicorn.run 启动服务器 # \'app:app\' 指示 uvicorn 在当前文件 (app.py) 中寻找名为 app 的应用实例 # host 和 port 指定服务器监听的 IP 和端口 # reload=True 启用热重载，方便开发过程中修改代码后自动重启 uvicorn.run(\'app:app\', host=\"127.0.0.1\", port=8000, reload=True)

三、文件上传

3.1、基础文件上传

3.1.1、小文件

from fastapi import FastAPI, File​app = FastAPI()​@app.post(\"/upload/\")def upload_file(file: bytes = File(...)): return {\"file_size\": len(file)} # 文件内容直接加载到内存

特点：

        文件以二进制形式读取，适合小于 10MB 的文件

        内存占用高，大文件可能导致崩溃

3.1.2、大小文件都可

from fastapi import UploadFile​@app.post(\"/upload/\")def upload_file(file: UploadFile): contents = await file.read() # 异步读取 return {\"filename\": file.filename, \"size\": len(contents)}

优势：

        自动处理内存和磁盘存储（超过阈值存磁盘）

        支持文件元数据（filename, content_type）

        提供异步文件操作方法（read(), write()）

3.2、多文件上传

from typing import List​@app.post(\"/batch-upload/\")def batch_upload(files: List[UploadFile] = File(...)): return {\"count\": len(files), \"names\": [f.filename for f in files]}

每个文件独立处理，避免内存溢出

3.3、文件保存到本地

3.3.1、同步保存（小文件）

@app.post(\"/save-file/\")async def save_file(file: UploadFile): with open(file.filename, \"wb\") as f: f.write(await file.read()) # 同步写入 return {\"status\": \"saved\"}

3.3.2、异步保存（大文件优化）

@app.post(\'/upload2/\')async def upload_file2(file:UploadFile): async with aiofiles.open(f\'./data/{file.filename}\',\'wb\') as f: chunk = await file.read(1024 * 1024 * 1024) while chunk: await f.write(chunk) chunk = await file.read(1024 * 1024 * 1024) return {\"status\": \"success\"}

 或

import aiofiles​@app.post(\"/save-large-file/\")async def save_large_file(file: UploadFile): async with aiofiles.open(file.filename, \"wb\") as f: while chunk := await file.read(1024 * 1024): # 分块读取1MB await f.write(chunk) return {\"status\": \"success\"}

最佳实践：

        使用 aiofiles 异步 IO 提升性能分块读取

        避免内存溢出

3.4、文件验证

from fastapi import HTTPException​ALLOWED_EXTENSIONS = {\".jpg\", \".jpeg\", \".gif\"}​@app.post(\"/upload-image/\")def upload_image(file: UploadFile): # 校验文件类型 ext = Path(file.filename).suffix.lower() if ext not in ALLOWED_EXTENSIONS: raise HTTPException(400, \"不支持的文件扩展名\")​ return {\"filename\": file.filename}

3.5、混合表单与文件上传

from fastapi import Form​@app.post(\"/submit-form/\")def submit_form( username: str = Form(...), avatar: UploadFile = File(...)): return {\"user\": username, \"avatar_size\": avatar.size}

四、请求对象Request

# 导入 FastAPI 核心类from fastapi import FastAPI, Request# 创建一个 FastAPI 应用实例app = FastAPI()# ---# 定义一个异步 POST 路由，用于处理文件上传（或其他 POST 请求）@app.post(\'/upload1/\')async def upload_file(request: Request): \"\"\" 处理 /upload1/ 路由的 POST 请求。 代码思路： - 通过在函数参数中声明 `request: Request`，FastAPI 会将当前的请求对象注入到 `request` 变量中。 - `Request` 对象提供了访问请求的各种属性和方法，例如 URL、方法、请求头、请求体等。 - 由于 `request.body()` 是一个异步操作（需要等待读取完整的请求体），所以函数需要声明为 `async def`。 实现功能： - 演示如何从 FastAPI 的 `Request` 对象中获取各种请求信息。 - 返回一个字典，其中包含了： - `请求URL`: 完整的请求 URL。 - `请求方法`: HTTP 请求方法（例如 GET, POST）。 - `请求头`: 所有的 HTTP 请求头，以字典形式呈现。 - `请求体`: 原始的请求体内容（需要 await request.body() 来获取）。 - `请求IP`: 客户端的 IP 地址。 - `请求路径`: 请求的路径部分（不包含查询参数）。 - `请求路径参数`: 路径中定义的参数（如果路由中有的话）。 - `请求查询参数`: URL 中的查询参数（例如 ?key=value）。 - `请求cookies`: 客户端发送的 Cookies。 \"\"\" # 获取请求体，这是一个异步操作，需要 await body_content = await request.body() return { \"请求URL\": str(request.url), # 将 URL 对象转换为字符串 \"请求方法\": request.method, \"请求头\": dict(request.headers), # 将 Headers 对象转换为字典 \"请求体\": body_content.decode(\'utf-8\') if body_content else \"无请求体\", # 解码请求体，如果存在的话 \"请求IP\": request.client.host if request.client else \"未知\", # 获取客户端IP，并处理可能为空的情况 \"请求路径\": request.scope.get(\'path\'), \"请求路径参数\": request.scope.get(\'path_params\'), \"请求查询参数\": dict(request.query_params), # 将 QueryParams 对象转换为字典 \"请求cookies\": dict(request.cookies) # 将 Cookies 对象转换为字典 }# ---# 运行服务器if __name__ == \'__main__\': # 导入 uvicorn 库，这是一个 ASGI 服务器 import uvicorn # 运行 FastAPI 应用 # app=\'demo12:app\' 表示在 demo12.py 文件中寻找名为 app 的应用实例 # host=\'127.0.0.1\' 表示监听本地回环地址 # port=8000 表示监听 8000 端口 # reload=True 表示文件修改后自动重启服务器，方便开发 uvicorn.run(app=\'demo12:app\', host=\'127.0.0.1\', port=8000, reload=True)

网络标签：

相关问题