#FastAPI学习笔记

大模型应用开发基础-FastAPI

学习内容:

• 后端基础: FastAPI 路由、异步 I/O、Pydantic 数据校验
• 要求：掌握 FastAPI 基础，能够创建路由、处理请求

解锁技能:

• 熟练使用 FastAPI 搭建 API

教程: FastAPI Official Tutorial

fastapi是一个用于搭建API的python web框架。

• 🟢web 部分:starlette是一个用于处理异步I/O的python异步框架。
• 🟢数据部分:Pydantic是一个用于数据校验的python库。

了解
ASGI（Asynchronous Server Gateway Interface，异步服务器网关接口） 是 Python 的一个标准接口规范，用于连接 Web 服务器和 Web 应用程序
协议转换：将网络上的 HTTP/WebSocket 请求字节流，解析并转换为 Python 应用能理解的异步事件（scope, receive, send），并把应用的响应发回给客户端。

一、环境搭建，跑通样例程序

python版本（FastAPI要求3.8+，这里用的3.11）

1. 安装依赖：(uvicorn是一个用于运行ASGI应用程序的服务器)

pip install fastapi uvicorn

2. 创建main.py官网样例程序

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
# 若声明异步函数，在def前加async关键字
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

3. 启动服务器
   uvicorn main:app --reload
   命令含义:

• main：main.py 文件（一个 Python “模块”）。
• app：在 main.py 文件中通过 app = FastAPI() 创建的对象。
• (可选)–reload：让服务器在更新代码后重新启动。仅在开发时使用该选项。
• (可选)–port 8001：指定服务器端口。

最好不要同时起两个cmd窗口用同一个端口执行该命令，改代码看效果容易搞混。

#idea终端窗口关了后，进程还在运行，需要结束进程
# 查看所有运行中的 python 进程
tasklist | findstr python

# 或者查找特定的 uvicorn 进程
tasklist | findstr uvicorn

# 结束指定进程（替换 PID 为实际进程 ID）
taskkill /PID <PID> /F

4. 验证

• 访问 http://127.0.0.1:8000 验证是否返回{“message”: “Hello FastAPI”}
• 查看自动生成的API文档，并尝试在文档界面调用API
  http://127.0.0.1:8000/docs
  http://127.0.0.1:8000/redoc
  

二、主要知识点

1. 路由基础（GET/POST）
   @app.get("/")
@app.post("/items/")

# 使用Pydantic模型定义请求体
from pydantic import BaseModel
# 定义 Pydantic 模型（数据校验+类型提示）
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool

@app.post("/items/")
def create_item(item: Item):
    return ![请添加图片描述](https://i-blog.csdnimg.cn/direct/5e0c39e8c49b417eac2e1b5baac51c97.png)
{
        "item_name": item.name,
        "item_price": item.price,
        "is_offer": item.is_offer
    }

2. Pydantic 数据校验
   class Xxx(BaseModel):这种形式声明的数据模型，会自动进行数据校验。
   扩展Item模型，添加校验规则：用到了邮箱格式校验，需要安装pydantic[email]包，安装命令pip install pydantic[email]
   启动：和之前一样uvicorn testPydantic:app --reload或直接运行文件

from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr

# 初始化 FastAPI 应用实例（核心，必须有）
app = FastAPI()

# 定义 Pydantic 模型，校验用户数据
class User(BaseModel):
    username: str = Field(min_length=3, max_length=20)  # 用户名长度限制 3-20
    email: EmailStr  # 自动校验邮箱格式
    age: int = Field(ge=0, le=120)  # 年龄范围 0-120

# 定义 POST 接口，接收并校验 User 数据
@app.post("/users/")
def create_user(user: User):
    # 将 Pydantic 模型实例序列化为 Python 字典
    return {"user_info": user.model_dump()}

# 启动服务的入口（可选，方便直接运行脚本）
if __name__ == "__main__":
    import uvicorn
    # 启动本地服务：地址 127.0.0.1，端口 8000，热重载（开发模式）
    uvicorn.run("testPydantic:app", host="127.0.0.1", port=8001, reload=True)

测试异常场景：

• 提交username=“ab”（长度不足）
• 提交email=“test”（非邮箱格式）
• 提交age=-1

自动校验效果：

• 若用户名长度 < 3（如 {“username”:“ab”,“email”:“test@example.com”,“age”:20}），接口会返回 422 错误，提示字段校验失败；
• 若邮箱格式错误（如 test@.com），也会自动返回校验错误

3. 异步 I/O

async 和 await

await 只能在 async def 定义的函数内部使用
有一个操作需要等待才能给出结果burgers = await get_burgers(2)
它告诉 Python 它必须等待 ⏸ get_burgers(2) 完成它的工作 🕙 ，然后将结果存储在 burgers 中

# testAsync.py
from fastapi import FastAPI
from pydantic import BaseModel
import asyncio
import time

# 初始化 FastAPI 应用
app = FastAPI(title="Async FastAPI Demo", version="1.0")

# 定义 Item 模型（Pydantic v2 推荐写法）
class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = False

# 异步 GET 接口（模拟异步I/O）
@app.get("/async/")
async def async_root():
    # 模拟异步操作（如异步数据库查询、异步HTTP请求）
    await asyncio.sleep(3)
    return {"message": "Hello Async FastAPI"}

# ❷ 同步接口（3秒延迟）
@app.get("/sync/")
def sync_root():
    time.sleep(3)  # 同步休眠：阻塞整个服务，其他请求排队等
    return {"message": "Hello Sync FastAPI"}

# 异步 POST 接口（使用 model_dump 替代 dict 方法）
@app.post("/async/items/")
async def async_create_item(item: Item):
    await asyncio.sleep(0.5)
    return {"item": item.model_dump()}

# 启动服务入口
if __name__ == "__main__":
    import uvicorn
    # 启动参数：指定端口、热重载（开发模式）
    uvicorn.run(
        "testAsync:app",  # 格式：文件名:应用实例名
        host="127.0.0.1",
        port=8000,
        reload=True
    )

高并发下异步性能更优的直观验证
用「同时发 10 个请求」测试（可通过浏览器 / Postman / 脚本模拟）：

接口类型	10个请求总耗时	核心原因
同步 /sync/	~30秒	每个请求阻塞3秒，10个请求串行执行（3×10）
异步 /async/	~3秒	所有请求同时挂起，3秒后一起返回（并行处理）

注意事项：

• 若在同步函数中使用 await 会报错，反之异步函数中也可调用同步代码（但会阻塞事件循环，需用 asyncio.to_thread 包装）；