一、表单数据

要使用表单，需预先安装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)