【AI】十八.大模型服务必备FastAPI Web框架

小滴课堂讲师 2025年09月21日 fastapi 预计阅读 14 分钟

#### Python Web框架Fast API技术介绍

##### 需求背景：现代Web开发痛点

* 传统框架（如Flask、Django）对异步支持弱，性能受限。
* 手动编写API文档耗时且易过时、缺乏自动化数据验证，易出现安全漏洞。

##### FastAPI是什么？

- 一个基于 Python 3.8+ 的现代 Web 框架，专为构建高性能 API 设计。
- 官方地址：
  - https://github.com/fastapi/fastapi
  - https://fastapi.tiangolo.com/
- 关键特性：
  - ⚡ 高性能：基于Starlette（异步）和Pydantic（数据验证）
  - 📚 自动文档：集成Swagger UI和Redoc，代码即文档。
  - ✅ 强类型安全：Python类型提示 + Pydantic模型，减少Bug。
  - 🚀 易学易用：代码简洁，适合快速开发API。
- 适用场景
  * 构建 RESTful API 或 大模型服务，微服务架构中的独立服务模块。
  * 需要自动生成 API 文档的项目。
  * **核心价值**：FastAPI = 高性能 + 强类型 + 自动文档。
- **优点**：
  - 开发效率高（代码量减少 40%+）
  - 严格的类型提示减少运行时错误。
- **缺点**：
  - 生态相对较新（插件和社区资源少于 Django/Flask）。
  - 学习曲线略高（需熟悉 异步编程）。

##### 什么是 Uvicorn（类似Java 生态的Tomcat）

* 是一个轻量级、高性能的 **ASGI（Asynchronous Server Gateway Interface）服务器** 
* 专门运行异步 Python Web 应用（如 FastAPI、Starlette 或 Django Channels）。
* 类似传统 WSGI 服务器 的异步升级版，支持 HTTP/1.1、WebSockets 和 ASGI 标准。
* **核心功能**
  * **异步处理**：基于 `asyncio` 库，原生支持 `async/await`，适合高并发场景（如实时通信、高频 API 调用）。
  * **高性能**：使用 C 语言编写的 `uvloop` 和 `httptools` 加速网络请求处理。
  * **自动重载**：开发时通过 `--reload` 参数监听代码变动，自动重启服务。
  * **协议支持**： HTTP/1.1 、WebSocket、实验性 HTTP/2（需额外配置）
* 与 FastAPI 的关系
  * 依赖关系：FastAPI 本身不包含服务器，需通过 Uvicorn（或其他 ASGI 服务器）启动。
  * 协同工作流程：
    * 用户通过 Uvicorn 启动 FastAPI 应用。
    * Uvicorn 监听 HTTP 请求，将请求转发给 FastAPI 处理。
    * FastAPI 处理请求并返回响应，由 Uvicorn 发送给客户端

##### 类似产品对比

#### FastAPI环境安装和基础案例实战

##### 环境安装

* FastAPI 依赖 Python 3.8 及更高版本, 安装fastapi版本 `0.115.12`
* 安装命令
  * 安装依赖库. `pip install "fastapi[standard]"`
  * 安装uvicorn服务器 `pip install "uvicorn[standard]"`

##### 案例实战

* 创建第一个Fast API应用

```python
  from typing import Union
  
  from fastapi import FastAPI
  
  # 创建FastAPI实例
  app = FastAPI() 
  
  #http请求方式类型：get、post、put、update、delete
  #不带参数访问路由
  @app.get("/")
  def read_root():
      return {"Hello": "World11"}
  
  
  # 带参数访问，比如 http://127.0.0.1:8000/items/5?q=xd
  # q参数通过 Union[str, None] 表示可以是字符串类型或空，这样就允许在请求中不提供 q 参数。
  @app.get("/items/{item_id}")
  def read_item(item_id: int, q: Union[str, None] = None):
      return {"item_id": item_id, "q": q}
  ```

* 启动服务

```python
  # 语法：uvicorn 文件的相对路径：实例名 --reload
  # 多级目录使用.,例如： uvicorn xx.xx.main:app -- reload
  
  #--reload：代码修改后自动重启（仅开发环境）
  # main:当前目录下的main.py文件，app：文件下的app实例变量
  uvicorn main:app --reload
  
  #指定程序监听端口
  uvicorn main:app --port 8002 --reload
  
  #下面这个方式也可以
  fastapi dev main.py
  ```

* 访问API与文档

* API地址：`http://localhost:8000`
  * 交互文档：`http://localhost:8000/docs`
  * 备用文档( 另一种交互式文档界面，具有清晰简洁的外观 )：`http://localhost:8000/redoc`

#### 进阶FastAPI Web多案例参数请求实战

##### 路由与 HTTP 方法

```python
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/{item_id}")
def read_item(item_id: int):  # 自动类型转换和验证
    return {"item_id": item_id}

@app.post("/items/")
def create_item(item: dict):
    return {"item": item}

@app.put("/items/{item_id}")
def update_item(item_id: int, item: dict):
    return {"item_id": item_id, "updated_item": item}

@app.delete("/items/{item_id}")
def delete_item(item_id: int):
    return {"status": "deleted", "item_id": item_id}

```

##### 请求头Header操作

```python
from fastapi import Header

@app.get("/header")
def read_item(item_id: int, token: str = Header("token")):
    return {"token": token,"item_id": item_id}
   
```

##### 获取多个请求头

```python
from fastapi import Request
@app.get("/all-headers/")
def get_all_headers(request: Request):
    return dict(request.headers)
```

##### 自定义响应头

```python
from fastapi.responses import JSONResponse
@app.get("/custom-header/")
def set_custom_header():
    content = {"message": "Hello World"}
    headers = {
        "X-Custom-Header": "my_value xdclass.net",
        "Cache-Control": "max-age=3600"
    }
    return JSONResponse(content=content, headers=headers)

```

##### 自定义响应码

```python
from fastapi import FastAPI, status,Response

@app.get("/status_code", status_code=200)
def create_item(name: str):
    if name == "Foo":
        return Response(status_code=status.HTTP_404_NOT_FOUND)
    return {"name": name}
```

#### FastAPI异步编程async-await多案例实战

##### 需求背景

* 什么是同步和异步？ 用一个奶茶店买饮料的场景帮理解同步和异步的区别

* 🧋 **同步（Synchronous）就像传统奶茶店**

```markdown
    你： 老板，我要一杯珍珠奶茶
    店员： 开始制作（5分钟）
    你： 干站着等，不能做其他事
    店员： 做好了！请取餐
    下一位顾客： 必须等前一个人完成才能点单
    特点： 必须按顺序处理，前一个任务没完成，后面全卡住
    ```

* 🚀 异步（Asynchronous）像智能奶茶店

```markdown
    你： 扫码下单珍珠奶茶
    系统： 收到订单（生成取餐号）
    你： 去旁边座位玩手机（不用干等）
    店员： 同时处理多个订单
    系统： 奶茶做好后叫号通知你
    特点： 下单后可以继续做其他事，系统并行处理多个任务
    ```

* 对应到代码中

```markdown
    # 同步代码（传统奶茶店模式）
    def make_tea():
        print("开始煮茶")    # 👨🍳 店员开始工作
        time.sleep(3)       # ⏳ 你干等着
        print("加珍珠")
        return "奶茶好了"
    
    # 异步代码（智能奶茶店模式）
    async def async_make_tea():
        print("开始煮茶")    
        await asyncio.sleep(3)  # 🚶♂️ 你可以去做其他事
        print("加珍珠")
        return "奶茶好了"
    ```

##### 同步和异步关键区别总结【伪代码】

* 同步版本（存在性能瓶颈）

```python
  import requests
  
  @app.get("/sync-news")
  def get_news_sync():
      # 顺序执行（总耗时=各请求之和）
      news1 = requests.get("https://api1.com").json()  # 2秒
      news2 = requests.get("https://api2.com").json()  # 2秒
      return {"total_time": 4}
  ```

* 异步版本（高效并发）

```python
  import httpx
  
  @app.get("/async-news")
  async def get_news_async():
      async with httpx.AsyncClient() as client:
          # 并行执行（总耗时≈最慢的请求）
          start = time.time()
          task1 = client.get("https://api1.com")
          task2 = client.get("https://api2.com")
          res1, res2 = await asyncio.gather(task1, task2)
          return {
              "total_time": time.time() - start  # ≈2秒
          }
  ```

##### 异步编程常见类库介绍

* * `asyncio`  类库

* Python 中用于编写**单线程并发代码**的库，基于**协程（Coroutine）**和**事件循环（Event Loop）**实现异步编程。

* 专为处理 I/O 密集型任务（如网络请求、文件读写、数据库操作）设计，提高程序的吞吐量和资源利用率

* **协程（Coroutine）**

- 使用 `async def` 定义的函数称为协程，返回一个协程对象，不会立即执行。

- 协程通过 `await` 关键字挂起自身，将控制权交还给事件循环，直到异步操作完成

```
        async def my_coroutine():
            await asyncio.sleep(1)
            print("Done!")
        ```

* **事件循环（Event Loop）**

- 事件循环是异步程序的核心，负责调度和执行协程、处理 I/O 事件、管理回调等。

- 通过 `asyncio.run()` 或手动创建事件循环来启动。

```python
        # 启动事件循环并运行协程
        asyncio.run(my_coroutine())
        ```

* **任务（Task）**

- 任务是对协程的封装，用于在事件循环中并发执行多个协程。

- 通过 `asyncio.create_task()` 创建任务

```python
        async def main():
            task = asyncio.create_task(my_coroutine())
            await task
        ```

* **Future**

- `Future` 是一个底层对象，表示异步操作的最终结果。通常开发者直接使用 `Task`（它是 `Future` 的子类）。

* `httpx`  类库

* Python 中一个现代化、功能丰富的 HTTP 客户端库，支持同步和异步请求

* `httpx` 是 `requests` 的现代替代品，结合了易用性与强大功能，尤其适合需要异步或 HTTP/2 支持的场景。

* 同步请求案例

```python
      import httpx
      # GET 请求
      response = httpx.get("https://httpbin.org/get")
      print(response.status_code)
      print(response.json())
      ```

* 异步请求案例

```python
      import httpx
      import asyncio
      
      async def fetch_data():
          async with httpx.AsyncClient() as client:
              response = await client.get("https://httpbin.org/get")
              print(response.json())
      
      asyncio.run(fetch_data())
      ```

* 使用客户端实例（推荐)

```python
      # 同步客户端
      with httpx.Client() as client:
          response = client.get("https://httpbin.org/get")
      
      # 异步客户端
      async with httpx.AsyncClient() as client:
          response = await client.get("https://httpbin.org/get")
      ```

##### 案例实战

* 基础同步和异步编程

```python
  from fastapi import FastAPI
  import asyncio
  import time
  
  app = FastAPI()
  
  # 同步路由执行（顺序阻塞）
  @app.get("/sync_func")
  def sync_endpoint():
      # 模拟耗时操作
      print("开始任务1")  # 立即执行
      time.sleep(3)      # 阻塞3秒
      print("开始任务2")  # 3秒后执行
      return {"status": "done"}
  
  # 异步异步执行（非阻塞切换）
  @app.get("/async_func")
  async def async_endpoint():  # 异步接口
      # 必须使用异步库
      print("开始任务A")   # 立即执行
      await asyncio.sleep(3)  # 释放控制权,异步等待
      print("开始任务B")   # 3秒后恢复执行
      return {"status": "done"}
  ```

* 综合案例实战：并发调用外部API

```python
  import asyncio
  import httpx
  from fastapi import FastAPI
  
  app = FastAPI()
  
  """
  异步函数：fetch_data
  功能：通过HTTP GET请求从指定URL获取数据并返回JSON格式的响应。
  参数：
      url (str): 目标API的URL地址。
  返回值：
      dict: 从目标URL获取的JSON格式数据。
  """
  async def fetch_data(url: str):
      async with httpx.AsyncClient() as client:
          response = await client.get(url)
          return response.json()
  
  
  
  """
  路由处理函数：get_news
  功能：定义一个FastAPI的GET路由，用于并发获取多个API的数据并返回整合后的结果。
  """
  @app.get("/xdclass")
  async def get_news():
      start = time.time()
      # 定义需要请求的API URL列表
      urls = [
          "https://api-v2.xdclass.net/api/funny/v1/get_funny",
          "https://api-v2.xdclass.net/api/banner/v1/list?location=home_top_ad",
          "https://api-v2.xdclass.net/api/rank/v1/hot_product",
          "http://localhost:8000/sync1",
          "http://localhost:8000/sync2",
      ]
      
      # 创建并发任务列表，每个任务调用fetch_data函数获取对应URL的数据
      tasks = [fetch_data(url) for url in urls]
      
      # 使用asyncio.gather并发执行所有任务，并等待所有任务完成
      results = await asyncio.gather(*tasks)
      print(f"共耗时 {time.time() - start} 秒")
      # 返回整合后的结果
      return results
  ```

* 常见错误模式演示

```python
  # 错误示例：在async函数中使用同步阻塞
  @app.get("/wrong")
  async def bad_example():
      time.sleep(5)  # 会阻塞整个事件循环！
  
  # 正确方案1：改用异步等待
  @app.get("/right1")
  async def good_example():
      await asyncio.sleep(5)
  ```

##### 最佳实践指南

* **必须使用async的三种场景**
    * 需要await调用的异步库操作
    * WebSocket通信端点
    * 需要后台任务处理的接口

* **以下情况适合同步路由**

- 纯CPU计算（如数学运算）
    - 使用同步数据库驱动
    - 快速返回的简单端点

* **路由声明规范**

* 所有路由优先使用`async def`
    * 仅在必须使用同步操作时用普通 `def`

* **性能优化建议**

- 保持async函数轻量化
    - 长时间CPU密集型任务使用后台线程
    - 使用连接池管理数据库/HTTP连接

#### 进阶FastAPI模型Pydantic 案例实战

##### 什么是数据模型

* 客户端能发送什么数据
* 服务端会返回什么数据
* 数据验证规则

##### 基础模型定义

```python
from pydantic import BaseModel

# 用户注册模型
class UserRegister(BaseModel):
    username: str          # 必填字段
    email: str | None = None  # 可选字段
    age: int = Field(18, gt=0)  # 带默认值和验证
```

##### 案例实战

* 请求体（POST/PUT数据）

```python
  from pydantic import BaseModel,Field
  class Product(BaseModel):
      name: str
      price: float = Field(..., gt=0)
      tags: list[str] = []
  
  @app.post("/products")
  async def create_product(product: Product):
      return product.model_dump()
      
      
  #测试数据
  {
      "name":"小滴",
      "price":111,
      "tags":["java","vue"]
  }
  ```

* 返回 Pydantic 模型

```python
  from pydantic import BaseModel
  class Item(BaseModel):
      name: str
      description: str = None
      price: float
      tax: float = None
  
  @app.post("/items")
  async def create_item(item: Item):
      return item
      
  #测试数据
  {
      "name":"小滴",
      "price":111,
      "description":"小滴课堂是在线充电平台"
  }
  ```

* 密码强度验证

```python
  from pydantic import BaseModel, field_validator
  
  class UserRegistration(BaseModel):
      username: str
      password: str
  
      @field_validator('password')
      def validate_password(cls, v):
          if len(v) < 8:
              raise ValueError('密码至少8位')
          if not any(c.isupper() for c in v):
              raise ValueError('必须包含大写字母')
          if not any(c.isdigit() for c in v):
              raise ValueError('必须包含数字')
          return v
  
  @app.post("/register/")
  async def register(user: UserRegistration):
      return {"username": user.username}
      
      
  #测试数据
  {
      "username":"小滴",
      "password":"123"
  }
  ```

评论区