1. FastAPI 简介为什么选择它FastAPI 是一个现代、快速高性能的 Web 框架用于基于 Python 3.6 标准类型提示构建 API。它由 Sebastián Ramírez 于 2018 年创建并迅速成为 Python 社区中最受欢迎的 Web 框架之一。其核心优势在于极致的性能基于 Starlette用于 Web 微服务和 Pydantic用于数据验证FastAPI 的性能可与 Node.js 和 Go 相媲美是目前最快的 Python Web 框架之一。开发效率高通过 Python 类型提示自动生成 API 文档Swagger UI 和 ReDoc并自动进行请求/响应数据的验证、序列化和文档化。易于学习设计直观减少了大量样板代码让开发者可以专注于业务逻辑。生产就绪内置对 WebSocket、GraphQL、CORS、依赖注入等现代 Web 功能的支持。根据 TechEmpower 基准测试FastAPI 在 JSON 序列化等场景下的性能远超 Flask 和 Django使其成为构建高性能微服务和 API 网关的理想选择。2. 环境搭建与第一个 FastAPI 应用2.1 安装 FastAPI首先确保你的 Python 版本在 3.6 以上然后使用 pip 安装 FastAPI 和 ASGI 服务器如 Uvicorn 或 Hypercornpip install fastapi pip install uvicorn[standard]2.2 创建第一个 API创建一个名为main.py的文件写入以下代码from fastapi import FastAPI 创建 FastAPI 实例 app FastAPI() 定义根路径路由 app.get(/) def read_root(): return {message: Hello, FastAPI!} 带路径参数的路由 app.get(/items/{item_id}) def read_item(item_id: int, q: str None): return {item_id: item_id, q: q}2.3 运行应用在终端中运行以下命令启动开发服务器uvicorn main:app --reload访问 http://127.0.0.1:8000 即可看到返回的 JSON 消息。访问 High Performance Web Crawler API - Swagger UI 可以查看自动生成的交互式 API 文档Swagger UI。3. 核心概念深度解析3.1 路径操作与 HTTP 方法FastAPI 使用装饰器来声明路径操作支持所有常见的 HTTP 方法from fastapi import FastAPI app FastAPI() app.get(/items/) def get_items(): return {method: GET} app.post(/items/) def create_item(): return {method: POST} app.put(/items/{item_id}) def update_item(item_id: int): return {method: PUT, item_id: item_id} app.delete(/items/{item_id}) def delete_item(item_id: int): return {method: DELETE, item_id: item_id} app.patch(/items/{item_id}) def partial_update_item(item_id: int): return {method: PATCH, item_id: item_id}3.2 路径参数与查询参数FastAPI 会自动从 URL 路径和查询字符串中提取参数并根据类型提示进行验证和转换。路径参数直接在路径中声明如/items/{item_id}。查询参数作为函数参数声明但不是路径的一部分可以有默认值。from typing import Optional from fastapi import FastAPI app FastAPI() app.get(/users/{user_id}/items/{item_id}) def read_user_item( user_id: int, item_id: str, q: Optional[str] None, short: bool False ): item {user_id: user_id, item_id: item_id} if q: item.update({q: q}) if not short: item.update({description: This is a long description}) return item3.3 请求体与 Pydantic 模型使用 Pydantic 模型来定义请求体的数据结构FastAPI 会自动验证、解析和生成文档。from pydantic import BaseModel from fastapi import FastAPI app FastAPI() 定义数据模型 class Item(BaseModel): name: str description: str None price: float tax: float None app.post(/items/) def create_item(item: Item): # item 已经是经过验证的 Pydantic 实例 item_dict item.dict() if item.tax: price_with_tax item.price item.tax item_dict.update({price_with_tax: price_with_tax}) return item_dict3.4 响应模型与状态码你可以使用response_model参数来指定返回数据的模型FastAPI 会自动过滤和序列化响应数据。同时可以使用status_code参数设置 HTTP 状态码。from typing import List from pydantic import BaseModel from fastapi import FastAPI, status app FastAPI() class ItemIn(BaseModel): name: str price: float class ItemOut(BaseModel): name: str price: float tax: float 10.0 app.post( /items/, response_modelItemOut, status_codestatus.HTTP_201_CREATED ) def create_item(item: ItemIn): # 业务逻辑处理 return ItemOut(**item.dict(), taxitem.price * 0.1)4. 高级特性与最佳实践4.1 依赖注入系统FastAPI 的依赖注入系统是其最强大的特性之一可以用于共享业务逻辑、数据库会话、认证检查等。from fastapi import Depends, FastAPI, HTTPException from typing import Optional app FastAPI() 简单的依赖函数 def common_parameters(q: Optional[str] None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) def read_items(commons: dict Depends(common_parameters)): return commons 类作为依赖项 class QueryParams: def init(self, q: Optional[str] None, skip: int 0, limit: int 100): self.q q self.skip skip self.limit limit app.get(/users/) def read_users(params: QueryParams Depends()): return params4.2 中间件与 CORSFastAPI 支持 Starlette 的中间件系统可以轻松添加跨域资源共享CORS、请求日志、认证等中间件。from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware app FastAPI() 添加 CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], # 允许的源 allow_credentialsTrue, allow_methods[], # 允许所有方法 allow_headers[], # 允许所有头 ) 自定义中间件示例 app.middleware(http) async def add_process_time_header(request, call_next): import time start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response4.3 后台任务与 WebSocketFastAPI 支持在响应返回后执行后台任务非常适合发送邮件、处理图片等耗时操作。同时它也原生支持 WebSocket 连接。from fastapi import FastAPI, BackgroundTasks, WebSocket import time app FastAPI() 后台任务 def write_log(message: str): with open(log.txt, modea) as log: log.write(f{time.time()}: {message}\n) app.post(/send-notification/) def send_notification(email: str, background_tasks: BackgroundTasks): background_tasks.add_task(write_log, fnotification sent to {email}) return {message: Notification sent in background} WebSocket 端点 app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data await websocket.receive_text() await websocket.send_text(fMessage text was: {data})4.4 数据库集成SQLAlchemy Alembic虽然 FastAPI 不强制使用任何特定的数据库工具但 SQLAlchemy 和 Alembic 是最常见的组合。# database.py from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker SQLALCHEMY_DATABASE_URL sqlite:///./test.db 或使用 PostgreSQL: postgresql://user:passwordlocalhost/dbname engine create_engine( SQLALCHEMY_DATABASE_URL, connect_args{check_same_thread: False} ) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() models.py from sqlalchemy import Column, Integer, String from database import Base class User(Base): tablename users id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue) hashed_password Column(String) is_active Column(Boolean, defaultTrue) crud.py 和 main.py 中的依赖注入 from sqlalchemy.orm import Session from fastapi import Depends from database import SessionLocal def get_db(): db SessionLocal() try: yield db finally: db.close() app.get(/users/{user_id}) def read_user(user_id: int, db: Session Depends(get_db)): user db.query(User).filter(User.id user_id).first() if user is None: raise HTTPException(status_code404, detailUser not found) return user5. 测试、部署与性能优化5.1 使用 Pytest 进行测试FastAPI 应用可以方便地使用 Pytest 进行测试包括路径操作、依赖项和数据库操作。# test_main.py from fastapi.testclient import TestClient from main import app client TestClient(app) def test_read_root(): response client.get(/) assert response.status_code 200 assert response.json() {message: Hello, FastAPI!} def test_create_item(): response client.post( /items/, json{name: Foo, price: 45.2}, ) assert response.status_code 201 assert name in response.json() assert response.json()[name] Foo5.2 部署到生产环境生产环境部署建议ASGI 服务器使用 Uvicorn 或 Hypercorn 配合 Gunicorn多进程。反向代理使用 Nginx 或 Traefik 作为反向代理处理 SSL/TLS、静态文件、负载均衡。进程管理使用 systemd、Supervisor 或 Docker 容器管理进程。Docker 部署示例# Dockerfile FROM python:3.9-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 80]5.3 性能优化技巧使用异步尽可能使用async def定义路径操作函数避免阻塞 I/O。连接池数据库连接使用连接池避免频繁创建和销毁连接。缓存对频繁读取且不常变化的数据使用 Redis 或内存缓存。静态文件使用 Nginx 或 CDN 服务静态文件减轻应用服务器压力。监控与日志集成 Prometheus、Grafana 进行监控使用结构化日志如 JSON 格式。6. 总结与学习资源FastAPI 以其出色的性能、开发体验和现代化特性已成为构建 Python Web API 的首选框架。本文涵盖了从基础到高级的核心概念但 FastAPI 的生态还在不断丰富中。推荐学习路径官方文档FastAPI 官方文档 是最全面、最新的学习资源。实战项目尝试构建一个完整的 CRUD 应用集成数据库、认证、文件上传等功能。社区资源关注 FastAPI 的 GitHub 仓库、Reddit 社区和 Discord 频道了解最新动态和最佳实践。进阶学习深入研究 Starlette 和 Pydantic 的源码理解 FastAPI 的内部工作原理。