当前位置： 首页 > ds >正文

FastAPI 入门指南

ds 2025/9/1 23:32:28

FastAPI 入门指南

FastAPI Logo

🚀 什么是 FastAPI？

FastAPI 是一个现代、快速的 Python Web 框架，专为构建 API 而设计。它由 Sebastián Ramírez（tiangolo）在 2018 年创建，旨在解决传统 Python Web 框架在 API 开发中的痛点。

为什么会有 FastAPI？

在 FastAPI 出现之前，Python 的 Web 开发主要依赖于：

  • Django：功能强大但相对厚重，适合全栈 Web 应用
  • Flask：轻量灵活但需要大量配置，缺乏现代特性
  • Tornado：支持异步但学习成本较高

开发者在构建现代 API 时面临的挑战：

  • 🐌 性能瓶颈：传统框架性能有限
  • 📝 文档维护：手动编写和维护 API 文档
  • 🔍 类型安全：缺乏强类型检查，容易产生运行时错误
  • ⏰ 开发效率：重复代码多，开发周期长

FastAPI 应运而生，旨在解决这些问题。

核心技术栈

FastAPI 构建在几个优秀的库之上：

组件作用优势
StarletteASGI 框架核心高性能异步处理
Pydantic数据验证和序列化基于类型提示的自动验证
uvloop事件循环比标准库快 2-4 倍
httptoolsHTTP 解析C 语言实现，极致性能

现代 Python 特性支持

FastAPI 充分利用了 Python 3.6+ 的现代特性：

🏷️ 类型提示（Type Hints）
# Python 3.6+ 类型提示
def get_user(user_id: int) -> dict:return {"id": user_id, "name": "John"}# FastAPI 利用类型提示自动验证
@app.get("/users/{user_id}")
async def get_user(user_id: int):  # 自动验证 user_id 为整数return {"id": user_id, "name": "John"}
⚡ 异步编程（async/await）
# 传统同步方式（阻塞）
def slow_operation():time.sleep(1)  # 阻塞整个进程return "done"# 现代异步方式（非阻塞）
async def fast_operation():await asyncio.sleep(1)  # 释放资源给其他请求return "done"

这些现代特性让 FastAPI 能够：

  • 🔍 编译时检查：IDE 能在编写代码时发现错误
  • 📈 高并发处理：异步特性支持大量并发请求
  • 🤖 自动化处理：减少手动配置和样板代码

核心优势

特性说明对比
🔥 高性能基于 Starlette 和 Pydantic性能可与 Node.js 相媲美
⚡ 快速开发自动生成文档和类型检查开发速度提升 200-300%
🛡️ 类型安全基于 Python 类型提示编译时发现错误
📚 自动文档无需配置的交互式 API 文档自动生成 OpenAPI/Swagger
🎯 易于学习直观的设计，符合直觉学习曲线平缓

设计理念

“快速编码，减少 bug，直观易用”

FastAPI 的设计遵循以下原则：

  • 标准优先：基于开放标准（OpenAPI、JSON Schema）
  • 类型驱动：利用 Python 类型提示
  • 性能至上：追求最佳性能表现
  • 开发体验：提供出色的 IDE 支持和调试体验

🏗️ RESTful API 设计基础

在学习 FastAPI 的具体用法之前，我们需要先了解 RESTful API 的设计原则。这是因为：

  1. FastAPI 专为 API 设计：了解好的 API 设计是必要基础
  2. 行业标准：RESTful 是当前 Web API 的主流设计风格
  3. 团队协作：统一的设计规范有助于团队协作
  4. FastAPI 内置支持：框架本身就是按 RESTful 原则设计的

什么是 REST？

REST（Representational State Transfer）是一种架构风格，不是协议或标准。它定义了一组约束条件，用于创建 Web 服务。

REST 核心原则

1. 资源导向
✅ 好的设计                    ❌ 不好的设计
GET /users                   GET /getUsers  
POST /users                  POST /createUser
PUT /users/123               PUT /updateUser/123
DELETE /users/123            DELETE /deleteUser/123
2. HTTP 方法映射
方法用途幂等性示例
GET获取资源✅GET /users
POST创建资源❌POST /users
PUT更新资源✅PUT /users/123
PATCH部分更新❌PATCH /users/123
DELETE删除资源✅DELETE /users/123
3. 状态码使用
  • 2xx 成功：200 OK, 201 Created, 204 No Content
  • 4xx 客户端错误：400 Bad Request, 404 Not Found, 409 Conflict
  • 5xx 服务器错误：500 Internal Server Error

API 设计实例

# 用户管理 API 设计
GET    /api/v1/users              # 获取用户列表
POST   /api/v1/users              # 创建新用户
GET    /api/v1/users/{id}         # 获取特定用户
PUT    /api/v1/users/{id}         # 更新用户
DELETE /api/v1/users/{id}         # 删除用户# 查询参数
GET /api/v1/users?page=1&limit=10&status=active

FastAPI 与 RESTful 的完美结合

现在我们已经了解了 RESTful API 的设计原则，接下来我们将看到 FastAPI 是如何让实现这些原则变得简单而优雅的：

  • 路径参数：@app.get("/users/{user_id}") 直接映射 REST 资源
  • HTTP 方法：装饰器直接对应 HTTP 动词（GET、POST、PUT、DELETE）
  • 状态码：自动或手动设置合适的 HTTP 状态码
  • 数据验证：Pydantic 模型自动验证请求和响应数据
  • 文档生成：自动生成符合 OpenAPI 规范的 API 文档

让我们从一个简单的 FastAPI 应用开始，逐步实现上面设计的用户管理 API。

⚡ 快速开始

环境准备

# 安装 FastAPI 和服务器
pip install "fastapi[all]"# 或分别安装
pip install fastapi uvicorn

第一个应用

# main.py
from fastapi import FastAPIapp = FastAPI(title="我的第一个 API",description="学习 FastAPI 的示例项目",version="1.0.0"
)@app.get("/")
async def root():return {"message": "Hello FastAPI!"}@app.get("/health")
async def health_check():return {"status": "healthy"}

运行应用

方式一：命令行启动
# 开发模式
uvicorn main:app --reload --port 8000# 生产模式
uvicorn main:app --workers 4 --host 0.0.0.0 --port 8000
方式二：代码内启动
import uvicornif __name__ == "__main__":uvicorn.run("main:app",host="127.0.0.1",port=8000,reload=True  # 开发模式)

Uvicorn 简介：

  • ASGI 服务器：支持异步处理，比传统 WSGI 更高效
  • 高性能：基于 uvloop 和 httptools
  • FastAPI 首选：官方推荐的运行方式

访问应用

  • 🌐 应用地址：http://localhost:8000
  • 📖 API 文档：http://localhost:8000/docs
  • 📚 备用文档：http://localhost:8000/redoc

🎯 核心概念

路径参数

from fastapi import Path
from enum import Enum# 基础路径参数
@app.get("/users/{user_id}")
async def get_user(user_id: int):return {"user_id": user_id}# 枚举验证
class ModelType(str, Enum):cnn = "cnn"rnn = "rnn"transformer = "transformer"@app.get("/models/{model_type}")
async def get_model(model_type: ModelType):return {"model": model_type}# 路径参数验证
@app.get("/items/{item_id}")
async def get_item(item_id: int = Path(..., gt=0, description="商品ID")
):return {"item_id": item_id}

查询参数

from fastapi import Query
from typing import Optional, List@app.get("/search")
async def search_items(q: str = Query(..., min_length=2, description="搜索关键词"),page: int = Query(1, ge=1, description="页码"),size: int = Query(10, ge=1, le=100, description="每页数量"),tags: List[str] = Query([], description="标签过滤")
):return {"query": q,"page": page,"size": size,"tags": tags}

请求头和 Cookie

from fastapi import Header, Cookie@app.get("/headers")
async def read_headers(user_agent: str = Header(None),accept_language: str = Header(None)
):return {"User-Agent": user_agent,"Accept-Language": accept_language}@app.get("/cookies")
async def read_cookies(session_id: str = Cookie(None)
):return {"session_id": session_id}

📦 数据处理

Pydantic 模型

from pydantic import BaseModel, Field, validator
from typing import Optional, List
from datetime import datetimeclass UserCreate(BaseModel):username: str = Field(..., min_length=3, max_length=20)email: str = Field(..., regex=r'^[\w\.-]+@[\w\.-]+\.\w+$')password: str = Field(..., min_length=6)age: int = Field(..., gt=0, le=120)class UserResponse(BaseModel):id: intusername: stremail: stris_active: bool = Truecreated_at: datetimeclass Config:orm_mode = True  # 支持 ORM 对象# 数据验证器
class Product(BaseModel):name: strprice: float = Field(..., gt=0)category: str@validator('category')def validate_category(cls, v):allowed = ['electronics', 'books', 'clothing']if v not in allowed:raise ValueError(f'类别必须是: {allowed}')return v

API 端点示例

# 创建用户
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):# 业务逻辑new_user = {"id": 1,"username": user.username,"email": user.email,"is_active": True,"created_at": datetime.now()}return new_user# 获取用户列表
@app.get("/users", response_model=List[UserResponse])
async def get_users(skip: int = 0, limit: int = 10
):# 模拟数据users = [{"id": 1,"username": "john","email": "john@example.com","is_active": True,"created_at": datetime.now()}]return users[skip:skip + limit]

完整CRUD示例

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from datetime import datetimeapp = FastAPI()# 数据模型
class UserCreate(BaseModel):username: stremail: strage: intclass UserUpdate(BaseModel):username: Optional[str] = Noneemail: Optional[str] = Noneage: Optional[int] = Noneclass User(BaseModel):id: intusername: stremail: strage: intcreated_at: datetimeis_active: bool = True# 模拟数据库
users_db = []
next_id = 1# CRUD 操作
@app.post("/users", response_model=User, status_code=201)
async def create_user(user: UserCreate):global next_idnew_user = User(id=next_id,username=user.username,email=user.email,age=user.age,created_at=datetime.now())users_db.append(new_user.dict())next_id += 1return new_user@app.get("/users", response_model=List[User])
async def get_users(skip: int = 0, limit: int = 10):return users_db[skip:skip + limit]@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int):user = next((u for u in users_db if u["id"] == user_id), None)if not user:raise HTTPException(status_code=404, detail="用户不存在")return user@app.put("/users/{user_id}", response_model=User)
async def update_user(user_id: int, user_update: UserUpdate):user = next((u for u in users_db if u["id"] == user_id), None)if not user:raise HTTPException(status_code=404, detail="用户不存在")# 更新字段for field, value in user_update.dict(exclude_unset=True).items():user[field] = valuereturn user@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):global users_dbusers_db = [u for u in users_db if u["id"] != user_id]

🔧 高级特性

依赖注入

from fastapi import Depends# 简单依赖
def get_current_user():return {"user": "john_doe"}@app.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):return current_user# 数据库依赖
def get_db():# 数据库连接逻辑db = "database_connection"try:yield dbfinally:# 清理逻辑pass@app.get("/users/{user_id}")
async def get_user(user_id: int, db = Depends(get_db)):return {"user_id": user_id, "db": "connected"}

背景任务

from fastapi import BackgroundTasks
import smtplibdef send_email(email: str, message: str):# 发送邮件的逻辑print(f"发送邮件到 {email}: {message}")@app.post("/send-notification/")
async def send_notification(email: str, background_tasks: BackgroundTasks
):background_tasks.add_task(send_email, email, "欢迎注册！")return {"message": "邮件将在后台发送"}

文件上传

from fastapi import File, UploadFile
from typing import List@app.post("/upload-file/")
async def upload_file(file: UploadFile = File(...)):return {"filename": file.filename,"content_type": file.content_type,"size": len(await file.read())}@app.post("/upload-multiple/")
async def upload_multiple(files: List[UploadFile] = File(...)):return [{"filename": file.filename, "size": len(await file.read())} for file in files]

中间件与CORS

from fastapi import Request
from fastapi.middleware.cors import CORSMiddleware
import time# CORS 中间件
app.add_middleware(CORSMiddleware,allow_origins=["http://localhost:3000"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 自定义中间件
@app.middleware("http")
async def add_process_time(request: Request, call_next):start_time = time.time()response = await call_next(request)process_time = time.time() - start_timeresponse.headers["X-Process-Time"] = str(process_time)return response

简单认证

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearersecurity = HTTPBearer()def verify_token(token: str = Depends(security)):# 简单的token验证if token.credentials != "valid-token":raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid token")return {"user": "authenticated_user"}@app.get("/protected")
async def protected_route(user: dict = Depends(verify_token)):return {"message": "Access granted", "user": user}

🗄️ 数据库集成

SQLAlchemy 基础设置

from sqlalchemy import create_engine, Column, Integer, String, Boolean
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, SessionDATABASE_URL = "sqlite:///./app.db"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()# 数据库模型
class User(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, index=True)username = Column(String, unique=True, index=True)email = Column(String, unique=True, index=True)is_active = Column(Boolean, default=True)Base.metadata.create_all(bind=engine)

CRUD 操作

def get_db():db = SessionLocal()try:yield dbfinally:db.close()@app.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):db_user = User(username=user.username, email=user.email)db.add(db_user)db.commit()db.refresh(db_user)return db_user@app.get("/users/{user_id}", response_model=UserResponse)
def read_user(user_id: int, db: Session = Depends(get_db)):user = db.query(User).filter(User.id == user_id).first()if not user:raise HTTPException(status_code=404, detail="User not found")return user

🧪 测试与部署

测试

# test_main.py
from fastapi.testclient import TestClient
from main import appclient = TestClient(app)def test_read_root():response = client.get("/")assert response.status_code == 200assert response.json() == {"message": "Hello FastAPI!"}def test_create_user():user_data = {"username": "testuser","email": "test@example.com","password": "password123","age": 25}response = client.post("/users", json=user_data)assert response.status_code == 201assert response.json()["username"] == "testuser"

部署

Docker 部署
FROM python:3.9WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txtCOPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
生产环境配置
# 生产环境启动
if __name__ == "__main__":uvicorn.run("main:app",host="0.0.0.0",port=8000,workers=4,  # 多进程access_log=True)

🎨 最佳实践

项目结构

app/
├── main.py              # 应用入口
├── models/              # 数据库模型
│   └── user.py
├── schemas/             # Pydantic 模型
│   └── user.py
├── routers/             # 路由模块
│   └── users.py
├── dependencies.py      # 依赖项
├── database.py         # 数据库配置
└── config.py           # 配置管理

路由组织

# routers/users.py
from fastapi import APIRouterrouter = APIRouter(prefix="/users", tags=["users"])@router.get("/")
async def get_users():return {"users": []}# main.py
from routers import usersapp.include_router(users.router, prefix="/api/v1")

配置管理

# config.py
from pydantic import BaseSettingsclass Settings(BaseSettings):app_name: str = "FastAPI App"database_url: strsecret_key: strclass Config:env_file = ".env"settings = Settings()

错误处理

from fastapi import HTTPException
from fastapi.responses import JSONResponse@app.exception_handler(404)
async def not_found_handler(request, exc):return JSONResponse(status_code=404,content={"detail": "Resource not found"})# 自定义异常
class CustomException(Exception):def __init__(self, message: str):self.message = message@app.exception_handler(CustomException)
async def custom_exception_handler(request, exc):return JSONResponse(status_code=400,content={"detail": exc.message})

实用技巧

1. 响应模型优化
from pydantic import BaseModel
from typing import Optionalclass UserResponse(BaseModel):id: intusername: stremail: strclass Config:# 排除None值exclude_none = True# 允许字段别名allow_population_by_field_name = True
2. 请求参数校验
from fastapi import Query, Path, Body@app.post("/users/{user_id}/items")
async def create_item(user_id: int = Path(..., gt=0, description="用户ID"),priority: int = Query(1, ge=1, le=5, description="优先级"),item: dict = Body(..., example={"name": "示例物品"})
):return {"user_id": user_id, "priority": priority, "item": item}
3. 日志配置
import logging# 配置日志
logging.basicConfig(level=logging.INFO,format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)logger = logging.getLogger(__name__)@app.middleware("http")
async def log_requests(request: Request, call_next):logger.info(f"请求: {request.method} {request.url}")response = await call_next(request)logger.info(f"响应: {response.status_code}")return response
4. API 版本管理
from fastapi import APIRouter# v1 路由
v1_router = APIRouter(prefix="/v1")
v1_router.include_router(users.router)# v2 路由
v2_router = APIRouter(prefix="/v2")
v2_router.include_router(users_v2.router)# 主应用
app.include_router(v1_router, prefix="/api")
app.include_router(v2_router, prefix="/api")
5. 缓存响应
from functools import lru_cache@lru_cache(maxsize=128)
def get_settings():return Settings()@app.get("/settings")
async def read_settings():return get_settings()

常见问题解决

1. CORS 错误
# 允许所有来源（仅开发环境）
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 生产环境应指定具体域名allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)
2. 大文件上传
from fastapi import Request@app.post("/upload-large/")
async def upload_large_file(request: Request):form = await request.form()file = form["file"]# 处理大文件的逻辑return {"filename": file.filename}
3. 数据库连接池
from sqlalchemy.pool import StaticPoolengine = create_engine(DATABASE_URL,poolclass=StaticPool,pool_size=20,max_overflow=0
)

🎯 总结

FastAPI 是现代 Python Web 开发的优秀选择：

核心优势

  • ⚡ 高性能：接近 Node.js 和 Go 的性能
  • 🛡️ 类型安全：基于 Python 类型提示
  • 📚 自动文档：无需配置的 API 文档
  • 🚀 快速开发：减少重复代码，提高效率
  • 🔧 现代特性：支持异步、依赖注入等

学习路径

  1. 基础概念 → 理解 REST 和 FastAPI 特性
  2. 核心功能 → 掌握路径参数、查询参数、数据模型
  3. 高级特性 → 学习依赖注入、中间件、安全
  4. 实际应用 → 数据库集成、测试、部署
  5. 最佳实践 → 项目结构、错误处理、性能优化

推荐资源

  • 📖 FastAPI 官方文档
  • 🔗 Pydantic 文档
  • 🐙 GitHub 仓库

开始你的 FastAPI 之旅吧！🚀

http://www.xdnf.cn/news/8555.html

相关文章：

  • 阿里通义千问3（Qwen3）开源掀桌子（附完整使用教程）
  • PET,Prompt Tuning,P Tuning,Lora,Qlora 大模型微调的简介
  • AI编程对传统软件开发的冲击和思考
  • 【Python数据库全栈指南】从SQL到ORM深度实践
  • youyu:91501
  • 互联网大厂Java求职面试：Spring Boot 3.2+自动配置原理、AOT编译及原生镜像
  • 嵌入式系统C语言编程常用设计模式---参数表驱动设计
  • 快排-P1923求第 k 小的数
  • 开发指南117-文字阴影特效
  • 七彩喜康养护理——科技赋能下的全周期健康守护
  • 60道Angular高频题整理(附答案背诵版)
  • 动态防御体系实战：AI如何重构DDoS攻防逻辑
  • 实时操作系统革命：实时Linux驱动的智能时代底层重构
  • 向量数据库该如何选择？Milvus 、ES、OpenSearch 快速对比：向量搜索能力与智能检索引擎的应用前景
  • 小白学习顺序表 之 通讯录实现
  • JAVA查漏补缺（2）
  • 并发容器（Collections）
  • 文章记单词 | 第109篇（六级）
  • 主成分分析基本概念及python代码使用
  • 【软件测试】第三章·软件测试基本方法（逻辑覆盖、路径覆盖）
  • 从数学融智学视域系统地理解《道德经》：38至56，德化社会
  • 【MySQL】实战时遇到的几个 tips
  • AAAI-2016《Approximate K-Means++ in Sublinear Time》
  • python实战:Python脚本后台运行的方法
  • docker部署并测试翻译模型-CSANMT连续语义增强机器翻译
  • 《Android 应用开发基础教程》——第十五章：Android 动画机制详解（属性动画、帧动画、过渡动画）
  • 深入理解SummaryWriter类与TensorBoard的基本使用
  • SurfaceFlinger及Android应用RenderThread角度观察Jank丢帧卡顿
  • 【漫话机器学习系列】274.基尼指数（Gini Index）
  • 在Vue3 + Vite 项目安装使用 Tailwind CSS 4.0报错