【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 发送给客户端 ##### 类似产品对比 | 框架 | 语言 | 特点 | 适用场景 | | :-------------- | :------ | :----------------------------------- | :---------------------- | | **Flask** | Python | 轻量级、灵活,但同步且无内置数据验证 | 小型应用、快速原型开发 | | **Django** | Python | 全功能(ORM、Admin),但重量级、同步 | 复杂 Web 应用(如 CMS) | | **Express** | Node.js | 高并发,但动态类型易出错 | JS 生态的 API 开发 | | **Spring Boot** | Java | 企业级功能,但配置复杂、启动慢 | 大型分布式系统 | | **FastAPI** | Python | 高性能、异步、类型安全、自动文档 | 高并发 API、微服务 | #### 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" } ```
评论区