1. 引言在当今的 Web 开发领域Python 凭借其简洁的语法、丰富的生态和强大的库支持已经成为构建后端服务的流行语言之一。而随着微服务架构和前后端分离模式的普及RESTful API 已成为服务间通信的标准范式。在众多 Python Web 框架中FastAPI 异军突起以其高性能、易用性和现代化的特性赢得了开发者的青睐。本文将带领读者深入探索如何使用 Python 和 FastAPI 构建高效、可扩展的 RESTful API 服务。从 REST 基础理论到 FastAPI 的核心特性从数据库集成到安全认证从测试部署到性能优化我们将通过大量代码示例和实践指南帮助读者掌握构建生产级 Web 服务的完整技能。无论你是初学者还是有一定经验的开发者都能从中获得实用价值。2. RESTful API 基础2.1 什么是 RESTRESTRepresentational State Transfer表现层状态转移是一种软件架构风格由 Roy Fielding 在其博士论文中提出。它不是一种标准而是一组约束条件用于指导分布式超媒体系统的设计。遵循 REST 原则的 Web 服务被称为 RESTful API。2.2 REST 的核心原则客户端-服务器分离用户界面与数据存储分离提高可移植性和可扩展性。无状态每个请求都包含服务器处理所需的所有信息服务器不保存客户端上下文。可缓存响应可被隐式或显式标记为可缓存提高性能。统一接口通过资源的标识URI、资源的表示JSON/XML和自描述消息实现统一交互。分层系统组件之间通过层级隔离客户端无需知道是否直接连接最终服务器。按需代码可选服务器可向客户端传输可执行代码如 JavaScript但实际较少使用。2.3 HTTP 方法详解RESTful API 利用 HTTP 方法表达对资源的操作GET获取资源的表示只读。POST创建新资源通常向父资源发送请求。PUT替换整个资源幂等。PATCH部分更新资源。DELETE删除资源。HEAD获取资源的元数据如响应头。OPTIONS获取资源支持的 HTTP 方法。2.4 状态码的意义合理的状态码能够让客户端准确理解请求结果2xx 成功200 OKGET/PUT 成功、201 CreatedPOST 成功、204 No ContentDELETE 成功。3xx 重定向较少在 API 中使用。4xx 客户端错误400 Bad Request请求格式错误、401 Unauthorized未认证、403 Forbidden无权限、404 Not Found资源不存在、422 Unprocessable Entity验证失败。5xx 服务器错误500 Internal Server Error、502 Bad Gateway 等。2.5 资源设计与 URI资源是 REST 的核心URI 应指向资源而非操作。设计建议使用名词复数表示资源集合/users、/orders。通过路径参数指定单个资源/users/{id}。使用嵌套表达关系/users/{id}/orders。避免动词用 HTTP 方法表达操作。使用查询参数过滤、排序、分页/users?page2limit10。3. FastAPI 简介FastAPI 是一个现代、快速高性能的 Python Web 框架基于 Starlette用于 Web 部分和 Pydantic用于数据部分。它于 2018 年由 Sebastián Ramírez 创建迅速成为最受欢迎的 Python 异步框架之一。3.1 FastAPI 的特点性能卓越与 Node.js 和 Go 相当是 Python 最快的框架之一基于 Starlette。开发效率高代码量减少 40% 以上通过类型提示自动生成交互式文档。自动生成 API 文档内置 Swagger UI 和 ReDoc支持 OpenAPI 规范。数据验证基于 Pydantic利用 Python 类型提示进行声明式验证减少手动检查。异步支持原生支持async/await可以处理高并发 I/O 操作。依赖注入强大的依赖注入系统简化可复用逻辑如数据库会话、认证。基于标准完全兼容 OpenAPI 和 JSON Schema易于与工具链集成。3.2 FastAPI 与其他框架对比框架特点适用场景Flask轻量、灵活、生态丰富但同步且缺少内置验证和文档小型项目、微服务、原型开发Django全栈框架自带 ORM、Admin、模板但较重同步大型应用、内容管理系统FastAPI高性能、异步、自动文档、类型提示现代特性RESTful API、微服务、实时应用4. 环境准备4.1 Python 版本要求FastAPI 需要 Python 3.7 及以上版本推荐 3.10以充分利用类型提示和异步特性。建议使用虚拟环境管理项目依赖。4.2 安装 FastAPI 和 Uvicornbash# 创建虚拟环境可选 python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows # 安装 FastAPI 和 ASGI 服务器 Uvicorn pip install fastapi uvicorn[standard]fastapi框架核心。uvicorn[standard]轻量级 ASGI 服务器支持自动重载。4.3 项目结构推荐对于中型及以上项目建议采用以下结构textmyproject/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── api/ # 路由模块 │ │ ├── __init__.py │ │ ├── v1/ # 版本管理 │ │ │ ├── __init__.py │ │ │ ├── endpoints/ │ │ │ │ ├── users.py │ │ │ │ └── items.py │ ├── core/ # 核心配置 │ │ ├── config.py │ │ └── security.py │ ├── models/ # 数据库模型 │ ├── schemas/ # Pydantic 模型 │ ├── crud/ # 数据库操作 │ ├── dependencies/ # 依赖项 │ └── utils/ # 工具函数 ├── tests/ # 测试 ├── .env # 环境变量 └── requirements.txt5. 第一个 FastAPI 应用5.1 Hello World创建一个main.py文件pythonfrom fastapi import FastAPI app FastAPI() app.get(/) async def root(): return {message: Hello World} app.get(/items/{item_id}) async def read_item(item_id: int, q: str None): return {item_id: item_id, q: q}5.2 运行与调试使用 Uvicorn 运行bashuvicorn main:app --reloadmain:appmain.py中的app实例。--reload开发模式代码修改后自动重启。打开浏览器访问http://127.0.0.1:8000看到 JSON 响应。自动生成的交互式文档位于http://127.0.0.1:8000/docsSwagger UI和http://127.0.0.1:8000/redoc。6. 路径操作与路由6.1 声明路径参数路径参数使用大括号{}声明并作为函数参数接收。FastAPI 会自动进行类型转换和验证。pythonapp.get(/users/{user_id}) async def get_user(user_id: int): return {user_id: user_id}若参数类型错误如传入字符串API 将返回 422 验证错误。6.2 查询参数查询参数是 URL 中?后的键值对通过函数参数声明。默认值为None的参数为可选否则为必需。pythonapp.get(/items/) async def list_items(skip: int 0, limit: int 10, category: str None): return {skip: skip, limit: limit, category: category}访问/items/?skip5limit20categorybook将返回相应参数。6.3 请求体与 Pydantic 模型对于 POST、PUT 等需要发送数据的请求使用 Pydantic 模型定义请求体结构。pythonfrom pydantic import BaseModel class Item(BaseModel): name: str price: float is_offer: bool False app.post(/items/) async def create_item(item: Item): return {item_name: item.name, price: item.price}FastAPI 会自动验证传入 JSON 是否符合模型定义并返回 422 错误。6.4 响应模型通过response_model参数指定响应数据的结构可用于过滤敏感字段如密码或保证输出格式。pythonfrom pydantic import BaseModel class ItemInDB(BaseModel): name: str price: float hashed_password: str class ItemOut(BaseModel): name: str price: float app.post(/items/, response_modelItemOut) async def create_item(item: ItemInDB): # 假设 item 包含 hashed_password但响应中不会返回 return itemFastAPI 会使用response_model进行过滤和序列化。7. 请求与响应深入7.1 Cookie 和 Header可以使用Cookie和Header函数从请求中提取相应信息。pythonfrom fastapi import Cookie, Header app.get(/cookies/) async def read_cookie(session_id: str Cookie(None)): return {session_id: session_id} app.get(/headers/) async def read_header(user_agent: str Header(None)): return {User-Agent: user_agent}注意Header 参数会自动将下划线转换为连字符如user_agent对应user-agent。7.2 表单数据与文件上传处理表单数据需安装python-multipartpip install python-multipart。pythonfrom fastapi import Form, File, UploadFile app.post(/login/) async def login(username: str Form(...), password: str Form(...)): return {username: username} app.post(/upload/) async def upload_file(file: UploadFile File(...)): contents await file.read() return {filename: file.filename, size: len(contents)}UploadFile提供异步读取接口适合大文件处理。7.3 自定义状态码和响应头通过status_code参数设置响应状态码通过Response对象或JSONResponse设置响应头。pythonfrom fastapi import FastAPI, status from fastapi.responses import JSONResponse app.post(/items/, status_codestatus.HTTP_201_CREATED) async def create_item(name: str): return {name: name} app.get(/custom-header/) async def custom_header(): content {message: Hello} headers {X-Custom-Header: value} return JSONResponse(contentcontent, headersheaders)8. 数据验证与序列化8.1 Pydantic 模型进阶Pydantic 模型不仅用于验证还支持类型转换、默认值、可选字段等。pythonfrom pydantic import BaseModel, Field from typing import Optional class User(BaseModel): id: int name: str email: str age: Optional[int] Field(None, ge0, le120) # 年龄范围 0-120 tags: list[str] []8.2 嵌套模型与复杂结构模型可以嵌套形成复杂数据结构。pythonclass Address(BaseModel): street: str city: str zip_code: str class UserWithAddress(BaseModel): id: int name: str address: Address8.3 自定义验证器通过validator装饰器添加自定义验证逻辑。pythonfrom pydantic import BaseModel, validator class User(BaseModel): name: str password1: str password2: str validator(name) def name_must_contain_space(cls, v): if not in v: raise ValueError(must contain a space) return v.title() validator(password2) def passwords_match(cls, v, values): if password1 in values and v ! values[password1]: raise ValueError(passwords do not match) return v9. 依赖注入系统9.1 什么是依赖注入依赖注入是一种设计模式将组件的依赖关系由外部传入而不是在组件内部创建。FastAPI 内置了简单而强大的依赖注入系统常用于数据库会话、认证、配置等共享逻辑。9.2 声明和使用依赖依赖可以是函数、类或可调用对象。通过Depends在路径操作函数中声明依赖。pythonfrom fastapi import Depends async def common_parameters(q: str None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) async def read_items(commons: dict Depends(common_parameters)): return commons app.get(/users/) async def read_users(commons: dict Depends(common_parameters)): return commons9.3 可调用依赖与全局依赖依赖可以是类实例也可以添加到整个应用或路由组。pythonclass CommonQueryParams: def __init__(self, q: str None, skip: int 0, limit: int 100): self.q q self.skip skip self.limit limit app.get(/items/) async def read_items(commons: CommonQueryParams Depends()): return commons全局依赖在FastAPI应用实例或APIRouter上使用dependencies参数。pythonapp FastAPI(dependencies[Depends(verify_token)])9.4 使用 DependsDepends可以嵌套依赖也可以依赖其他依赖形成依赖图。FastAPI 会自动处理缓存在同一个请求内多次调用同一依赖会共享结果。10. 数据库集成10.1 选择 ORMSQLAlchemy 与 Tortoise-ORMSQLAlchemy功能强大支持同步和异步1.4生态成熟适合复杂查询。Tortoise-ORM原生异步API 类似 Django ORM轻量易用。本文以 SQLAlchemy 异步模式为例。安装依赖bashpip install sqlalchemy[asyncio] asyncpg # PostgreSQL10.2 使用 SQLAlchemy 定义模型python# app/models/user.py from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue, nullableFalse) hashed_password Column(String, nullableFalse) is_active Column(Integer, default1)10.3 异步数据库操作创建数据库引擎和会话python# app/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker DATABASE_URL postgresqlasyncpg://user:passlocalhost/dbname engine create_async_engine(DATABASE_URL, echoTrue) AsyncSessionLocal sessionmaker(engine, class_AsyncSession, expire_on_commitFalse) async def get_db() - AsyncSession: async with AsyncSessionLocal() as session: yield session10.4 CRUD 示例在路径操作中使用依赖获取数据库会话python# app/api/endpoints/users.py from fastapi import Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.database import get_db from app.models.user import User from app.schemas.user import UserCreate, UserOut app.post(/users/, response_modelUserOut) async def create_user(user: UserCreate, db: AsyncSession Depends(get_db)): # 检查邮箱是否已存在 result await db.execute(select(User).where(User.email user.email)) existing result.scalar_one_or_none() if existing: raise HTTPException(status_code400, detailEmail already registered) # 创建新用户密码哈希见后 db_user User(emailuser.email, hashed_passwordfakehash) db.add(db_user) await db.commit() await db.refresh(db_user) return db_user11. 用户认证与授权11.1 OAuth2 与 JWT 基础OAuth2 是一个授权框架FastAPI 内置了对 OAuth2 密码流的支持。JWTJSON Web Token是一种轻量级的令牌格式常用于 API 认证。安装依赖bashpip install python-jose[cryptography] passlib[bcrypt] python-multipart11.2 密码哈希与验证使用passlib进行密码哈希pythonfrom passlib.context import CryptContext pwd_context CryptContext(schemes[bcrypt], deprecatedauto) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password)11.3 实现登录接口与 Token 颁发python# app/core/security.py from datetime import datetime, timedelta from jose import JWTError, jwt SECRET_KEY your-secret-key ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 def create_access_token(data: dict, expires_delta: timedelta None): to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt登录接口python# app/api/endpoints/auth.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.core.security import verify_password, create_access_token from app.models.user import User oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) app.post(/token) async def login(form_data: OAuth2PasswordRequestForm Depends(), db: AsyncSession Depends(get_db)): user await authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException(status_code400, detailIncorrect username or password) access_token create_access_token(data{sub: user.email}) return {access_token: access_token, token_type: bearer} async def authenticate_user(db: AsyncSession, email: str, password: str): result await db.execute(select(User).where(User.email email)) user result.scalar_one_or_none() if not user or not verify_password(password, user.hashed_password): return False return user11.4 依赖注入获取当前用户创建一个依赖项从请求中提取 Token 并验证用户。python# app/api/deps.py from fastapi import Depends, HTTPException, status from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.core.security import SECRET_KEY, ALGORITHM from app.models.user import User async def get_current_user(token: str Depends(oauth2_scheme), db: AsyncSession Depends(get_db)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould not validate credentials, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) email: str payload.get(sub) if email is None: raise credentials_exception except JWTError: raise credentials_exception result await db.execute(select(User).where(User.email email)) user result.scalar_one_or_none() if user is None: raise credentials_exception return user async def get_current_active_user(current_user: User Depends(get_current_user)): if not current_user.is_active: raise HTTPException(status_code400, detailInactive user) return current_user然后在需要保护的路由中使用pythonapp.get(/users/me) async def read_users_me(current_user: User Depends(get_current_active_user)): return current_user11.5 角色与权限控制可以通过在用户模型中添加role字段然后在依赖中检查权限。pythondef require_admin(current_user: User Depends(get_current_active_user)): if current_user.role ! admin: raise HTTPException(status_code403, detailNot enough permissions) return current_user app.delete(/users/{user_id}) async def delete_user(user_id: int, admin: User Depends(require_admin)): # 删除逻辑 return {message: deleted}12. 中间件与 CORS12.1 自定义中间件中间件可以在每个请求处理前后执行逻辑。创建中间件使用app.middleware(http)。pythonimport time from fastapi import Request app.middleware(http) async def add_process_time_header(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response12.2 处理 CORS跨域资源共享CORS是浏览器安全策略需要后端允许特定来源访问。使用CORSMiddleware配置。pythonfrom fastapi.middleware.cors import CORSMiddleware origins [ http://localhost:3000, https://myfrontend.com, ] app.add_middleware( CORSMiddleware, allow_originsorigins, allow_credentialsTrue, allow_methods[*], allow_headers[*], )13. 错误处理与异常13.1 HTTPException 的使用在路径操作中当出现业务错误时可以抛出HTTPException。pythonfrom fastapi import HTTPException app.get(/items/{item_id}) async def read_item(item_id: int): if item_id not in items: raise HTTPException(status_code404, detailItem not found) return {item: items[item_id]}13.2 自定义异常处理器可以注册自定义异常处理器统一处理特定异常。pythonfrom fastapi import FastAPI, Request from fastapi.responses import JSONResponse class UnicornException(Exception): def __init__(self, name: str): self.name name app.exception_handler(UnicornException) async def unicorn_exception_handler(request: Request, exc: UnicornException): return JSONResponse( status_code418, content{message: fOops! {exc.name} did something.}, ) app.get(/unicorns/{name}) async def read_unicorn(name: str): if name yolo: raise UnicornException(namename) return {unicorn_name: name}14. 测试 FastAPI 应用14.1 使用 TestClientFastAPI 提供了TestClient基于 Requests 库用于测试同步路径操作。对于异步路径操作TestClient 内部会处理。安装pytest和httpxbashpip install pytest httpx14.2 编写单元测试创建test_main.pypythonfrom fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_read_root(): response client.get(/) assert response.status_code 200 assert response.json() {message: Hello World} def test_create_item(): response client.post(/items/, json{name: Foo, price: 10.5}) assert response.status_code 200 assert response.json()[name] Foo14.3 异步测试对于依赖数据库等异步操作的测试可以使用pytest-asyncio和异步客户端。pythonimport pytest from httpx import AsyncClient from app.main import app pytest.mark.asyncio async def test_async_endpoint(): async with AsyncClient(appapp, base_urlhttp://test) as ac: response await ac.get(/async-endpoint) assert response.status_code 20015. 部署与性能15.1 使用 Uvicorn 与 Gunicorn生产环境通常使用 Gunicorn 作为进程管理器Uvicorn 作为 worker。bashpip install gunicorn uvicorn gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000-wworker 进程数一般设置为 CPU 核心数 ×2 1。-kworker 类型UvicornWorker 支持 ASGI。15.2 Docker 化部署编写DockerfiledockerfileFROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY ./app ./app CMD [gunicorn, app.main:app, -w, 4, -k, uvicorn.workers.UvicornWorker, --bind, 0.0.0.0:8000]15.3 环境变量配置敏感信息如数据库密码、密钥应通过环境变量传入。使用pydantic-settings管理配置。安装pip install pydantic-settingspythonfrom pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str sqliteaiosqlite:///./test.db secret_key: str changeme algorithm: str HS256 access_token_expire_minutes: int 30 class Config: env_file .env settings Settings()15.4 性能调优建议使用异步数据库驱动如 asyncpg、aiomysql。启用响应压缩pip install fastapi[compress]添加GZipMiddleware。合理使用缓存如 Redis 缓存热点数据。避免同步阻塞操作使用asyncio.to_thread将同步任务提交到线程池。数据库连接池调优SQLAlchemy 的池大小配置。使用反向代理如 Nginx处理静态文件和负载均衡。16. 高级主题16.1 WebSocket 支持FastAPI 原生支持 WebSocket适合实时应用。pythonfrom fastapi import 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})16.2 后台任务对于耗时操作可以使用BackgroundTasks将其放在响应后执行不阻塞客户端。pythonfrom fastapi import BackgroundTasks def write_log(message: str): with open(log.txt, a) as f: f.write(message) app.post(/send-notification) async def send_notification(background_tasks: BackgroundTasks, email: str): background_tasks.add_task(write_log, fnotification sent to {email}) return {message: Notification sent in background}16.3 APIRouter 模块化使用APIRouter将路由分组便于管理。python# app/api/v1/endpoints/users.py from fastapi import APIRouter router APIRouter(prefix/users, tags[users]) router.get(/) async def read_users(): return [{name: Alice}]在main.py中引入pythonfrom app.api.v1.endpoints import users app.include_router(users.router)16.4 元数据与文档定制可以修改 OpenAPI 文档的标题、描述等。pythonapp FastAPI( titleMy API, descriptionThis is a very fancy API, version2.5.0, openapi_tags[{name: users, description: Operations with users}] )17. 最佳实践与项目结构17.1 大型应用组织模式按模块拆分将功能相关的路由、模型、schema 放在同一目录如users模块。使用 APIRouter 版本管理/api/v1/...便于后期升级。分层架构表现层路由→ 业务逻辑层service→ 数据访问层repository/crud。17.2 分层架构示例textapp/ ├── api/ # 路由层 ├── services/ # 业务逻辑 ├── repositories/ # 数据库操作 ├── models/ # SQLAlchemy 模型 └── schemas/ # Pydantic 模型例如用户创建流程路由接收请求调用UserService.create_user。Service 调用UserRepository.create并处理业务规则如密码哈希。Repository 执行数据库插入。17.3 配置管理使用pydantic-settings从环境变量和.env文件加载配置确保不同环境开发、测试、生产配置分离。python# app/core/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str Awesome API admin_email: str database_url: str redis_url: str redis://localhost:6379/0 class Config: env_file .env settings Settings()
Python, FastAPI 与 RESTful API：构建高效 Web 服务的指南
发布时间：2026/7/25 17:11:11
1. 引言在当今的 Web 开发领域Python 凭借其简洁的语法、丰富的生态和强大的库支持已经成为构建后端服务的流行语言之一。而随着微服务架构和前后端分离模式的普及RESTful API 已成为服务间通信的标准范式。在众多 Python Web 框架中FastAPI 异军突起以其高性能、易用性和现代化的特性赢得了开发者的青睐。本文将带领读者深入探索如何使用 Python 和 FastAPI 构建高效、可扩展的 RESTful API 服务。从 REST 基础理论到 FastAPI 的核心特性从数据库集成到安全认证从测试部署到性能优化我们将通过大量代码示例和实践指南帮助读者掌握构建生产级 Web 服务的完整技能。无论你是初学者还是有一定经验的开发者都能从中获得实用价值。2. RESTful API 基础2.1 什么是 RESTRESTRepresentational State Transfer表现层状态转移是一种软件架构风格由 Roy Fielding 在其博士论文中提出。它不是一种标准而是一组约束条件用于指导分布式超媒体系统的设计。遵循 REST 原则的 Web 服务被称为 RESTful API。2.2 REST 的核心原则客户端-服务器分离用户界面与数据存储分离提高可移植性和可扩展性。无状态每个请求都包含服务器处理所需的所有信息服务器不保存客户端上下文。可缓存响应可被隐式或显式标记为可缓存提高性能。统一接口通过资源的标识URI、资源的表示JSON/XML和自描述消息实现统一交互。分层系统组件之间通过层级隔离客户端无需知道是否直接连接最终服务器。按需代码可选服务器可向客户端传输可执行代码如 JavaScript但实际较少使用。2.3 HTTP 方法详解RESTful API 利用 HTTP 方法表达对资源的操作GET获取资源的表示只读。POST创建新资源通常向父资源发送请求。PUT替换整个资源幂等。PATCH部分更新资源。DELETE删除资源。HEAD获取资源的元数据如响应头。OPTIONS获取资源支持的 HTTP 方法。2.4 状态码的意义合理的状态码能够让客户端准确理解请求结果2xx 成功200 OKGET/PUT 成功、201 CreatedPOST 成功、204 No ContentDELETE 成功。3xx 重定向较少在 API 中使用。4xx 客户端错误400 Bad Request请求格式错误、401 Unauthorized未认证、403 Forbidden无权限、404 Not Found资源不存在、422 Unprocessable Entity验证失败。5xx 服务器错误500 Internal Server Error、502 Bad Gateway 等。2.5 资源设计与 URI资源是 REST 的核心URI 应指向资源而非操作。设计建议使用名词复数表示资源集合/users、/orders。通过路径参数指定单个资源/users/{id}。使用嵌套表达关系/users/{id}/orders。避免动词用 HTTP 方法表达操作。使用查询参数过滤、排序、分页/users?page2limit10。3. FastAPI 简介FastAPI 是一个现代、快速高性能的 Python Web 框架基于 Starlette用于 Web 部分和 Pydantic用于数据部分。它于 2018 年由 Sebastián Ramírez 创建迅速成为最受欢迎的 Python 异步框架之一。3.1 FastAPI 的特点性能卓越与 Node.js 和 Go 相当是 Python 最快的框架之一基于 Starlette。开发效率高代码量减少 40% 以上通过类型提示自动生成交互式文档。自动生成 API 文档内置 Swagger UI 和 ReDoc支持 OpenAPI 规范。数据验证基于 Pydantic利用 Python 类型提示进行声明式验证减少手动检查。异步支持原生支持async/await可以处理高并发 I/O 操作。依赖注入强大的依赖注入系统简化可复用逻辑如数据库会话、认证。基于标准完全兼容 OpenAPI 和 JSON Schema易于与工具链集成。3.2 FastAPI 与其他框架对比框架特点适用场景Flask轻量、灵活、生态丰富但同步且缺少内置验证和文档小型项目、微服务、原型开发Django全栈框架自带 ORM、Admin、模板但较重同步大型应用、内容管理系统FastAPI高性能、异步、自动文档、类型提示现代特性RESTful API、微服务、实时应用4. 环境准备4.1 Python 版本要求FastAPI 需要 Python 3.7 及以上版本推荐 3.10以充分利用类型提示和异步特性。建议使用虚拟环境管理项目依赖。4.2 安装 FastAPI 和 Uvicornbash# 创建虚拟环境可选 python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows # 安装 FastAPI 和 ASGI 服务器 Uvicorn pip install fastapi uvicorn[standard]fastapi框架核心。uvicorn[standard]轻量级 ASGI 服务器支持自动重载。4.3 项目结构推荐对于中型及以上项目建议采用以下结构textmyproject/ ├── app/ │ ├── __init__.py │ ├── main.py # 应用入口 │ ├── api/ # 路由模块 │ │ ├── __init__.py │ │ ├── v1/ # 版本管理 │ │ │ ├── __init__.py │ │ │ ├── endpoints/ │ │ │ │ ├── users.py │ │ │ │ └── items.py │ ├── core/ # 核心配置 │ │ ├── config.py │ │ └── security.py │ ├── models/ # 数据库模型 │ ├── schemas/ # Pydantic 模型 │ ├── crud/ # 数据库操作 │ ├── dependencies/ # 依赖项 │ └── utils/ # 工具函数 ├── tests/ # 测试 ├── .env # 环境变量 └── requirements.txt5. 第一个 FastAPI 应用5.1 Hello World创建一个main.py文件pythonfrom fastapi import FastAPI app FastAPI() app.get(/) async def root(): return {message: Hello World} app.get(/items/{item_id}) async def read_item(item_id: int, q: str None): return {item_id: item_id, q: q}5.2 运行与调试使用 Uvicorn 运行bashuvicorn main:app --reloadmain:appmain.py中的app实例。--reload开发模式代码修改后自动重启。打开浏览器访问http://127.0.0.1:8000看到 JSON 响应。自动生成的交互式文档位于http://127.0.0.1:8000/docsSwagger UI和http://127.0.0.1:8000/redoc。6. 路径操作与路由6.1 声明路径参数路径参数使用大括号{}声明并作为函数参数接收。FastAPI 会自动进行类型转换和验证。pythonapp.get(/users/{user_id}) async def get_user(user_id: int): return {user_id: user_id}若参数类型错误如传入字符串API 将返回 422 验证错误。6.2 查询参数查询参数是 URL 中?后的键值对通过函数参数声明。默认值为None的参数为可选否则为必需。pythonapp.get(/items/) async def list_items(skip: int 0, limit: int 10, category: str None): return {skip: skip, limit: limit, category: category}访问/items/?skip5limit20categorybook将返回相应参数。6.3 请求体与 Pydantic 模型对于 POST、PUT 等需要发送数据的请求使用 Pydantic 模型定义请求体结构。pythonfrom pydantic import BaseModel class Item(BaseModel): name: str price: float is_offer: bool False app.post(/items/) async def create_item(item: Item): return {item_name: item.name, price: item.price}FastAPI 会自动验证传入 JSON 是否符合模型定义并返回 422 错误。6.4 响应模型通过response_model参数指定响应数据的结构可用于过滤敏感字段如密码或保证输出格式。pythonfrom pydantic import BaseModel class ItemInDB(BaseModel): name: str price: float hashed_password: str class ItemOut(BaseModel): name: str price: float app.post(/items/, response_modelItemOut) async def create_item(item: ItemInDB): # 假设 item 包含 hashed_password但响应中不会返回 return itemFastAPI 会使用response_model进行过滤和序列化。7. 请求与响应深入7.1 Cookie 和 Header可以使用Cookie和Header函数从请求中提取相应信息。pythonfrom fastapi import Cookie, Header app.get(/cookies/) async def read_cookie(session_id: str Cookie(None)): return {session_id: session_id} app.get(/headers/) async def read_header(user_agent: str Header(None)): return {User-Agent: user_agent}注意Header 参数会自动将下划线转换为连字符如user_agent对应user-agent。7.2 表单数据与文件上传处理表单数据需安装python-multipartpip install python-multipart。pythonfrom fastapi import Form, File, UploadFile app.post(/login/) async def login(username: str Form(...), password: str Form(...)): return {username: username} app.post(/upload/) async def upload_file(file: UploadFile File(...)): contents await file.read() return {filename: file.filename, size: len(contents)}UploadFile提供异步读取接口适合大文件处理。7.3 自定义状态码和响应头通过status_code参数设置响应状态码通过Response对象或JSONResponse设置响应头。pythonfrom fastapi import FastAPI, status from fastapi.responses import JSONResponse app.post(/items/, status_codestatus.HTTP_201_CREATED) async def create_item(name: str): return {name: name} app.get(/custom-header/) async def custom_header(): content {message: Hello} headers {X-Custom-Header: value} return JSONResponse(contentcontent, headersheaders)8. 数据验证与序列化8.1 Pydantic 模型进阶Pydantic 模型不仅用于验证还支持类型转换、默认值、可选字段等。pythonfrom pydantic import BaseModel, Field from typing import Optional class User(BaseModel): id: int name: str email: str age: Optional[int] Field(None, ge0, le120) # 年龄范围 0-120 tags: list[str] []8.2 嵌套模型与复杂结构模型可以嵌套形成复杂数据结构。pythonclass Address(BaseModel): street: str city: str zip_code: str class UserWithAddress(BaseModel): id: int name: str address: Address8.3 自定义验证器通过validator装饰器添加自定义验证逻辑。pythonfrom pydantic import BaseModel, validator class User(BaseModel): name: str password1: str password2: str validator(name) def name_must_contain_space(cls, v): if not in v: raise ValueError(must contain a space) return v.title() validator(password2) def passwords_match(cls, v, values): if password1 in values and v ! values[password1]: raise ValueError(passwords do not match) return v9. 依赖注入系统9.1 什么是依赖注入依赖注入是一种设计模式将组件的依赖关系由外部传入而不是在组件内部创建。FastAPI 内置了简单而强大的依赖注入系统常用于数据库会话、认证、配置等共享逻辑。9.2 声明和使用依赖依赖可以是函数、类或可调用对象。通过Depends在路径操作函数中声明依赖。pythonfrom fastapi import Depends async def common_parameters(q: str None, skip: int 0, limit: int 100): return {q: q, skip: skip, limit: limit} app.get(/items/) async def read_items(commons: dict Depends(common_parameters)): return commons app.get(/users/) async def read_users(commons: dict Depends(common_parameters)): return commons9.3 可调用依赖与全局依赖依赖可以是类实例也可以添加到整个应用或路由组。pythonclass CommonQueryParams: def __init__(self, q: str None, skip: int 0, limit: int 100): self.q q self.skip skip self.limit limit app.get(/items/) async def read_items(commons: CommonQueryParams Depends()): return commons全局依赖在FastAPI应用实例或APIRouter上使用dependencies参数。pythonapp FastAPI(dependencies[Depends(verify_token)])9.4 使用 DependsDepends可以嵌套依赖也可以依赖其他依赖形成依赖图。FastAPI 会自动处理缓存在同一个请求内多次调用同一依赖会共享结果。10. 数据库集成10.1 选择 ORMSQLAlchemy 与 Tortoise-ORMSQLAlchemy功能强大支持同步和异步1.4生态成熟适合复杂查询。Tortoise-ORM原生异步API 类似 Django ORM轻量易用。本文以 SQLAlchemy 异步模式为例。安装依赖bashpip install sqlalchemy[asyncio] asyncpg # PostgreSQL10.2 使用 SQLAlchemy 定义模型python# app/models/user.py from sqlalchemy import Column, Integer, String from sqlalchemy.ext.declarative import declarative_base Base declarative_base() class User(Base): __tablename__ users id Column(Integer, primary_keyTrue, indexTrue) email Column(String, uniqueTrue, indexTrue, nullableFalse) hashed_password Column(String, nullableFalse) is_active Column(Integer, default1)10.3 异步数据库操作创建数据库引擎和会话python# app/database.py from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker DATABASE_URL postgresqlasyncpg://user:passlocalhost/dbname engine create_async_engine(DATABASE_URL, echoTrue) AsyncSessionLocal sessionmaker(engine, class_AsyncSession, expire_on_commitFalse) async def get_db() - AsyncSession: async with AsyncSessionLocal() as session: yield session10.4 CRUD 示例在路径操作中使用依赖获取数据库会话python# app/api/endpoints/users.py from fastapi import Depends, HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.database import get_db from app.models.user import User from app.schemas.user import UserCreate, UserOut app.post(/users/, response_modelUserOut) async def create_user(user: UserCreate, db: AsyncSession Depends(get_db)): # 检查邮箱是否已存在 result await db.execute(select(User).where(User.email user.email)) existing result.scalar_one_or_none() if existing: raise HTTPException(status_code400, detailEmail already registered) # 创建新用户密码哈希见后 db_user User(emailuser.email, hashed_passwordfakehash) db.add(db_user) await db.commit() await db.refresh(db_user) return db_user11. 用户认证与授权11.1 OAuth2 与 JWT 基础OAuth2 是一个授权框架FastAPI 内置了对 OAuth2 密码流的支持。JWTJSON Web Token是一种轻量级的令牌格式常用于 API 认证。安装依赖bashpip install python-jose[cryptography] passlib[bcrypt] python-multipart11.2 密码哈希与验证使用passlib进行密码哈希pythonfrom passlib.context import CryptContext pwd_context CryptContext(schemes[bcrypt], deprecatedauto) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password): return pwd_context.hash(password)11.3 实现登录接口与 Token 颁发python# app/core/security.py from datetime import datetime, timedelta from jose import JWTError, jwt SECRET_KEY your-secret-key ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 def create_access_token(data: dict, expires_delta: timedelta None): to_encode data.copy() if expires_delta: expire datetime.utcnow() expires_delta else: expire datetime.utcnow() timedelta(minutesACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt登录接口python# app/api/endpoints/auth.py from fastapi import Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.core.security import verify_password, create_access_token from app.models.user import User oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) app.post(/token) async def login(form_data: OAuth2PasswordRequestForm Depends(), db: AsyncSession Depends(get_db)): user await authenticate_user(db, form_data.username, form_data.password) if not user: raise HTTPException(status_code400, detailIncorrect username or password) access_token create_access_token(data{sub: user.email}) return {access_token: access_token, token_type: bearer} async def authenticate_user(db: AsyncSession, email: str, password: str): result await db.execute(select(User).where(User.email email)) user result.scalar_one_or_none() if not user or not verify_password(password, user.hashed_password): return False return user11.4 依赖注入获取当前用户创建一个依赖项从请求中提取 Token 并验证用户。python# app/api/deps.py from fastapi import Depends, HTTPException, status from jose import JWTError, jwt from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db from app.core.security import SECRET_KEY, ALGORITHM from app.models.user import User async def get_current_user(token: str Depends(oauth2_scheme), db: AsyncSession Depends(get_db)): credentials_exception HTTPException( status_codestatus.HTTP_401_UNAUTHORIZED, detailCould not validate credentials, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) email: str payload.get(sub) if email is None: raise credentials_exception except JWTError: raise credentials_exception result await db.execute(select(User).where(User.email email)) user result.scalar_one_or_none() if user is None: raise credentials_exception return user async def get_current_active_user(current_user: User Depends(get_current_user)): if not current_user.is_active: raise HTTPException(status_code400, detailInactive user) return current_user然后在需要保护的路由中使用pythonapp.get(/users/me) async def read_users_me(current_user: User Depends(get_current_active_user)): return current_user11.5 角色与权限控制可以通过在用户模型中添加role字段然后在依赖中检查权限。pythondef require_admin(current_user: User Depends(get_current_active_user)): if current_user.role ! admin: raise HTTPException(status_code403, detailNot enough permissions) return current_user app.delete(/users/{user_id}) async def delete_user(user_id: int, admin: User Depends(require_admin)): # 删除逻辑 return {message: deleted}12. 中间件与 CORS12.1 自定义中间件中间件可以在每个请求处理前后执行逻辑。创建中间件使用app.middleware(http)。pythonimport time from fastapi import Request app.middleware(http) async def add_process_time_header(request: Request, call_next): start_time time.time() response await call_next(request) process_time time.time() - start_time response.headers[X-Process-Time] str(process_time) return response12.2 处理 CORS跨域资源共享CORS是浏览器安全策略需要后端允许特定来源访问。使用CORSMiddleware配置。pythonfrom fastapi.middleware.cors import CORSMiddleware origins [ http://localhost:3000, https://myfrontend.com, ] app.add_middleware( CORSMiddleware, allow_originsorigins, allow_credentialsTrue, allow_methods[*], allow_headers[*], )13. 错误处理与异常13.1 HTTPException 的使用在路径操作中当出现业务错误时可以抛出HTTPException。pythonfrom fastapi import HTTPException app.get(/items/{item_id}) async def read_item(item_id: int): if item_id not in items: raise HTTPException(status_code404, detailItem not found) return {item: items[item_id]}13.2 自定义异常处理器可以注册自定义异常处理器统一处理特定异常。pythonfrom fastapi import FastAPI, Request from fastapi.responses import JSONResponse class UnicornException(Exception): def __init__(self, name: str): self.name name app.exception_handler(UnicornException) async def unicorn_exception_handler(request: Request, exc: UnicornException): return JSONResponse( status_code418, content{message: fOops! {exc.name} did something.}, ) app.get(/unicorns/{name}) async def read_unicorn(name: str): if name yolo: raise UnicornException(namename) return {unicorn_name: name}14. 测试 FastAPI 应用14.1 使用 TestClientFastAPI 提供了TestClient基于 Requests 库用于测试同步路径操作。对于异步路径操作TestClient 内部会处理。安装pytest和httpxbashpip install pytest httpx14.2 编写单元测试创建test_main.pypythonfrom fastapi.testclient import TestClient from app.main import app client TestClient(app) def test_read_root(): response client.get(/) assert response.status_code 200 assert response.json() {message: Hello World} def test_create_item(): response client.post(/items/, json{name: Foo, price: 10.5}) assert response.status_code 200 assert response.json()[name] Foo14.3 异步测试对于依赖数据库等异步操作的测试可以使用pytest-asyncio和异步客户端。pythonimport pytest from httpx import AsyncClient from app.main import app pytest.mark.asyncio async def test_async_endpoint(): async with AsyncClient(appapp, base_urlhttp://test) as ac: response await ac.get(/async-endpoint) assert response.status_code 20015. 部署与性能15.1 使用 Uvicorn 与 Gunicorn生产环境通常使用 Gunicorn 作为进程管理器Uvicorn 作为 worker。bashpip install gunicorn uvicorn gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000-wworker 进程数一般设置为 CPU 核心数 ×2 1。-kworker 类型UvicornWorker 支持 ASGI。15.2 Docker 化部署编写DockerfiledockerfileFROM python:3.10-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY ./app ./app CMD [gunicorn, app.main:app, -w, 4, -k, uvicorn.workers.UvicornWorker, --bind, 0.0.0.0:8000]15.3 环境变量配置敏感信息如数据库密码、密钥应通过环境变量传入。使用pydantic-settings管理配置。安装pip install pydantic-settingspythonfrom pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str sqliteaiosqlite:///./test.db secret_key: str changeme algorithm: str HS256 access_token_expire_minutes: int 30 class Config: env_file .env settings Settings()15.4 性能调优建议使用异步数据库驱动如 asyncpg、aiomysql。启用响应压缩pip install fastapi[compress]添加GZipMiddleware。合理使用缓存如 Redis 缓存热点数据。避免同步阻塞操作使用asyncio.to_thread将同步任务提交到线程池。数据库连接池调优SQLAlchemy 的池大小配置。使用反向代理如 Nginx处理静态文件和负载均衡。16. 高级主题16.1 WebSocket 支持FastAPI 原生支持 WebSocket适合实时应用。pythonfrom fastapi import 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})16.2 后台任务对于耗时操作可以使用BackgroundTasks将其放在响应后执行不阻塞客户端。pythonfrom fastapi import BackgroundTasks def write_log(message: str): with open(log.txt, a) as f: f.write(message) app.post(/send-notification) async def send_notification(background_tasks: BackgroundTasks, email: str): background_tasks.add_task(write_log, fnotification sent to {email}) return {message: Notification sent in background}16.3 APIRouter 模块化使用APIRouter将路由分组便于管理。python# app/api/v1/endpoints/users.py from fastapi import APIRouter router APIRouter(prefix/users, tags[users]) router.get(/) async def read_users(): return [{name: Alice}]在main.py中引入pythonfrom app.api.v1.endpoints import users app.include_router(users.router)16.4 元数据与文档定制可以修改 OpenAPI 文档的标题、描述等。pythonapp FastAPI( titleMy API, descriptionThis is a very fancy API, version2.5.0, openapi_tags[{name: users, description: Operations with users}] )17. 最佳实践与项目结构17.1 大型应用组织模式按模块拆分将功能相关的路由、模型、schema 放在同一目录如users模块。使用 APIRouter 版本管理/api/v1/...便于后期升级。分层架构表现层路由→ 业务逻辑层service→ 数据访问层repository/crud。17.2 分层架构示例textapp/ ├── api/ # 路由层 ├── services/ # 业务逻辑 ├── repositories/ # 数据库操作 ├── models/ # SQLAlchemy 模型 └── schemas/ # Pydantic 模型例如用户创建流程路由接收请求调用UserService.create_user。Service 调用UserRepository.create并处理业务规则如密码哈希。Repository 执行数据库插入。17.3 配置管理使用pydantic-settings从环境变量和.env文件加载配置确保不同环境开发、测试、生产配置分离。python# app/core/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): app_name: str Awesome API admin_email: str database_url: str redis_url: str redis://localhost:6379/0 class Config: env_file .env settings Settings()
