FastAPI 是一个用于构建 API 的现代、快速高性能的 Web 框架使用 Python 并基于标准的 Python 类型提示。如果你正在寻找一个能够显著提升开发效率、减少代码错误、并且自带完整文档的 Python Web 框架FastAPI 绝对值得一试。这个框架最大的特点是快——不仅运行速度快开发速度也快。它基于 Python 类型提示让你在编写代码时就能获得编辑器的自动补全和类型检查大大减少了调试时间。更重要的是FastAPI 会自动生成交互式 API 文档支持 Swagger UI 和 ReDoc 两种界面让你的 API 测试和文档维护变得异常简单。本文将从零开始带你掌握 FastAPI 的核心用法包括环境搭建、基础 API 开发、数据验证、依赖注入、数据库集成等实战内容。无论你是 Python 初学者还是有经验的开发者都能快速上手并应用到实际项目中。1. FastAPI 核心能力速览能力项说明项目类型Python Web 框架专注于 API 开发性能表现极高性能可与 NodeJS 和 Go 并肩最快的 Python 框架之一开发效率功能开发速度提升约 200% ~ 300%人为错误减少约 40%学习曲线易于学习和使用基于标准的 Python 类型提示文档支持自动生成交互式 API 文档Swagger UI ReDoc标准化基于 OpenAPI 和 JSON Schema 开放标准异步支持原生支持 async/await适合高并发场景生产就绪被 Microsoft、Uber、Netflix、Cisco 等公司用于生产环境启动方式命令行启动、FastAPI CLI、Uvicorn 服务器适合场景REST API 开发、微服务、机器学习服务接口、快速原型开发2. FastAPI 适用场景与使用边界FastAPI 特别适合以下场景推荐使用场景需要快速开发 RESTful API 的后端服务构建微服务架构中的各个服务组件为机器学习模型提供预测接口服务需要自动生成 API 文档的团队协作项目高并发要求的实时数据处理服务需要严格数据验证的金融或电商应用不推荐场景传统的服务端渲染网页应用建议使用 Django 或 Flask 配合模板超大型单体应用FastAPI 更适合微服务架构只需要简单静态页面的项目技术边界提醒FastAPI 专注于 API 开发前端界面需要配合其他框架对于复杂的后台管理界面建议使用专门的 Admin 框架文件上传等功能需要配置合适的中间件和存储方案3. 环境准备与前置条件在开始 FastAPI 之旅前确保你的开发环境满足以下要求3.1 系统要求操作系统Windows 10/11、macOS 10.14、LinuxUbuntu 16.04、CentOS 7Python 版本Python 3.7推荐 Python 3.8 以获得最佳体验内存至少 4GB RAM开发环境生产环境建议 8GB磁盘空间至少 500MB 可用空间3.2 开发工具准备推荐使用以下工具组合# 代码编辑器/IDE - VS Code推荐 Python 扩展 - PyCharm专业版或社区版 - Vim/Neovim 相关插件 # 版本控制 - Git # 包管理 - pipPython 自带 - 可选poetry 或 pipenv3.3 Python 环境检查在终端中运行以下命令检查当前环境# 检查 Python 版本 python --version # 或 python3 --version # 检查 pip 版本 pip --version # 如果上述命令不工作尝试 py --version # Windows如果 Python 版本低于 3.7需要先升级 Python。推荐使用 pyenv 或直接下载最新版本。4. FastAPI 安装与项目初始化4.1 创建虚拟环境虚拟环境是 Python 开发的最佳实践可以隔离项目依赖# 创建项目目录 mkdir fastapi-tutorial cd fastapi-tutorial # 创建虚拟环境方法一使用 venv python -m venv venv # 激活虚拟环境 # Windows venv\Scripts\activate # Linux/macOS source venv/bin/activate # 创建虚拟环境方法二使用 conda conda create -n fastapi-env python3.9 conda activate fastapi-env4.2 安装 FastAPI使用 pip 安装 FastAPI 及其标准依赖# 安装 FastAPI 和标准依赖推荐 pip install fastapi[standard] # 或者只安装核心功能 pip install fastapi # 验证安装 python -c import fastapi; print(fastapi.__version__)标准依赖包包含uvicornASGI 服务器用于运行 FastAPI 应用pydantic数据验证和设置管理starlette轻量级 ASGI 框架/工具包4.3 创建第一个 FastAPI 应用创建一个简单的main.py文件from fastapi import FastAPI # 创建 FastAPI 实例 app FastAPI() # 定义根路径路由 app.get(/) async def read_root(): return {message: 欢迎使用 FastAPI, status: 运行正常} # 带路径参数的路由 app.get(/items/{item_id}) async def read_item(item_id: int, q: str None): return {item_id: item_id, query: q} # 带请求体的 POST 路由 app.post(/items/) async def create_item(item: dict): return {received_item: item}5. 启动与测试 FastAPI 应用5.1 使用 FastAPI CLI 启动FastAPI 提供了便捷的命令行工具# 开发模式启动推荐 fastapi dev main.py # 生产模式启动 fastapi run main.py启动后你会看到类似输出╭────────── FastAPI CLI - Development mode ───────────╮ │ │ │ Serving at: http://127.0.0.1:8000 │ │ │ │ API docs: http://127.0.0.1:8000/docs │ │ │ │ Running in development mode, for production use: │ │ │ │ fastapi run │ │ │ ╰─────────────────────────────────────────────────────╯5.2 使用 Uvicorn 直接启动也可以直接使用 Uvicorn 服务器# 开发模式支持热重载 uvicorn main:app --reload --host 0.0.0.0 --port 8000 # 生产模式 uvicorn main:app --host 0.0.0.0 --port 80005.3 测试 API 接口启动服务后可以通过多种方式测试浏览器测试访问http://127.0.0.1:8000/查看根路径访问http://127.0.0.1:8000/items/42?qtest测试路径参数和查询参数命令行测试curl# 测试 GET 请求 curl http://127.0.0.1:8000/ # 测试带参数的 GET 请求 curl http://127.0.0.1:8000/items/123?qhello # 测试 POST 请求 curl -X POST http://127.0.0.1:8000/items/ \ -H Content-Type: application/json \ -d {name: 新项目, price: 99.99}Python 代码测试import requests # 测试根路径 response requests.get(http://127.0.0.1:8000/) print(response.json()) # 测试物品接口 response requests.get(http://127.0.0.1:8000/items/456?qtest) print(response.json())6. 自动生成的交互式文档FastAPI 最强大的功能之一就是自动生成 API 文档6.1 Swagger UI 文档访问http://127.0.0.1:8000/docs可以看到完整的交互式文档界面查看所有 API 端点直接测试每个接口查看请求/响应模型查看状态码和错误信息6.2 ReDoc 文档访问http://127.0.0.1:8000/redoc可以看到更简洁的文档界面更适合阅读和分享清晰的参数说明响应示例展示6.3 OpenAPI 规范访问http://127.0.0.1:8000/openapi.json可以获取原始的 OpenAPI 规范文件可用于代码生成工具API 测试工具集成第三方文档系统7. 数据验证与 Pydantic 模型FastAPI 使用 Pydantic 进行数据验证这是其核心优势之一7.1 创建数据模型from pydantic import BaseModel from typing import Optional, List from datetime import datetime class Item(BaseModel): name: str description: Optional[str] None price: float tax: Optional[float] None tags: List[str] [] class User(BaseModel): username: str email: str full_name: Optional[str] None class Order(BaseModel): id: int items: List[Item] user: User created_at: datetime datetime.now()7.2 在路由中使用模型from fastapi import FastAPI from pydantic import BaseModel app FastAPI() class Item(BaseModel): name: str price: float is_offer: bool None app.post(/items/) async def create_item(item: Item): # FastAPI 会自动验证请求体是否符合 Item 模型 return { item_name: item.name, item_price: item.price, is_offer: item.is_offer } app.put(/items/{item_id}) async def update_item(item_id: int, item: Item, q: str None): return { item_id: item_id, item_name: item.name, item_price: item.price, query: q }7.3 高级数据验证Pydantic 支持丰富的数据验证功能from pydantic import BaseModel, Field, EmailStr, validator from typing import List class UserCreate(BaseModel): username: str Field(..., min_length3, max_length50, regex^[a-zA-Z0-9_]$) email: EmailStr age: int Field(..., ge0, le150) password: str Field(..., min_length8) tags: List[str] Field(default_factorylist) validator(username) def username_alphanumeric(cls, v): if not v.replace(_, ).isalnum(): raise ValueError(用户名只能包含字母、数字和下划线) return v validator(password) def password_strength(cls, v): if not any(c.isupper() for c in v): raise ValueError(密码必须包含至少一个大写字母) if not any(c.isdigit() for c in v): raise ValueError(密码必须包含至少一个数字) return v8. 路径参数、查询参数与请求体8.1 路径参数app.get(/users/{user_id}) async def read_user(user_id: int): return {user_id: user_id} app.get(/products/{category}/{product_id}) async def read_product(category: str, product_id: int): return {category: category, product_id: product_id}8.2 查询参数from typing import Optional, List app.get(/items/) async def read_items( skip: int 0, # 必需参数有默认值 limit: int 10, # 必需参数有默认值 q: Optional[str] None, # 可选参数 tags: List[str] [] # 列表参数 ): return { skip: skip, limit: limit, query: q, tags: tags }8.3 请求体参数from pydantic import BaseModel class Item(BaseModel): name: str description: Optional[str] None price: float tax: Optional[float] None app.put(/items/{item_id}) async def update_item(item_id: int, item: Item): return {item_id: item_id, **item.dict()}8.4 混合使用各种参数app.put(/products/{product_id}) async def update_product( product_id: int, # 路径参数 item: Item, # 请求体 q: Optional[str] None, # 查询参数 short: bool False # 查询参数布尔类型 ): result {product_id: product_id, **item.dict()} if q: result.update({query: q}) if not short: result.update({description: 完整描述信息}) return result9. 响应模型与状态码处理9.1 响应模型from pydantic import BaseModel class UserResponse(BaseModel): id: int username: str email: str is_active: bool class ItemResponse(BaseModel): id: int name: str price: float owner: UserResponse app.post(/users/, response_modelUserResponse) async def create_user(user: UserCreate): # 返回的数据会自动按照 UserResponse 模型进行验证和过滤 return user_data app.get(/items/{item_id}, response_modelItemResponse) async def read_item(item_id: int): # 只返回 response_model 中定义的字段 return item_data9.2 状态码处理from fastapi import FastAPI, HTTPException, status app.post(/items/, status_codestatus.HTTP_201_CREATED) async def create_item(item: Item): # 创建成功返回 201 return {message: 物品创建成功, item: item} app.get(/items/{item_id}) async def read_item(item_id: int): if item_id 100: # 物品不存在返回 404 raise HTTPException( status_code404, detail物品不存在, headers{X-Error: 物品ID超出范围} ) return {item_id: item_id} app.put(/items/{item_id}) async def update_item(item_id: int, item: Item): if item_id 0: # 参数错误返回 422 raise HTTPException(status_code422, detail物品ID不能为0) return {item_id: item_id, updated_item: item}10. 依赖注入系统FastAPI 的依赖注入系统是其另一个强大功能10.1 基本依赖from fastapi import Depends, FastAPI app FastAPI() # 简单的依赖函数 def common_parameters(q: str None, skip: int 0, limit: int 10): 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 commons10.2 类作为依赖class Pagination: def __init__(self, skip: int 0, limit: int 10): self.skip skip self.limit limit class Database: def __init__(self): self.connection 模拟数据库连接 def get_users(self, pagination: Pagination): return f获取用户列表跳过 {pagination.skip}限制 {pagination.limit} def get_database(): return Database() def get_pagination(skip: int 0, limit: int 10) - Pagination: return Pagination(skipskip, limitlimit) app.get(/users/) async def read_users( db: Database Depends(get_database), pagination: Pagination Depends(get_pagination) ): return db.get_users(pagination)10.3 带缓存的依赖from fastapi import Depends import time class Cache: def __init__(self): self.data {} self.expire_time 300 # 5分钟 def get(self, key): if key in self.data and time.time() - self.data[key][timestamp] self.expire_time: return self.data[key][value] return None def set(self, key, value): self.data[key] {value: value, timestamp: time.time()} cache Cache() def get_cache(): return cache def get_user_data(user_id: int, cache: Cache Depends(get_cache)): # 先尝试从缓存获取 cached_data cache.get(fuser_{user_id}) if cached_data: return {source: cache, data: cached_data} # 缓存中没有模拟从数据库获取 user_data f用户{user_id}的详细信息 cache.set(fuser_{user_id}, user_data) return {source: database, data: user_data} app.get(/users/{user_id}) async def read_user(user_data: dict Depends(get_user_data)): return user_data11. 数据库集成与 ORM11.1 SQLAlchemy 集成from sqlalchemy import create_engine, Column, Integer, String, Float from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker from pydantic import BaseModel # 数据库配置 SQLALCHEMY_DATABASE_URL sqlite:///./test.db engine create_engine(SQLALCHEMY_DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) Base declarative_base() # 数据库模型 class ItemModel(Base): __tablename__ items id Column(Integer, primary_keyTrue, indexTrue) name Column(String, indexTrue) description Column(String, indexTrue) price Column(Float) # Pydantic 模型 class Item(BaseModel): name: str description: str None price: float class Config: orm_mode True # 依赖项获取数据库会话 def get_db(): db SessionLocal() try: yield db finally: db.close() app.post(/items/, response_modelItem) async def create_item(item: Item, db Depends(get_db)): db_item ItemModel(**item.dict()) db.add(db_item) db.commit() db.refresh(db_item) return db_item app.get(/items/{item_id}, response_modelItem) async def read_item(item_id: int, db Depends(get_db)): item db.query(ItemModel).filter(ItemModel.id item_id).first() if item is None: raise HTTPException(status_code404, detail物品未找到) return item11.2 异步数据库支持from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker from sqlalchemy.ext.declarative import declarative_base # 异步数据库配置 SQLALCHEMY_DATABASE_URL sqliteaiosqlite:///./test.db engine create_async_engine(SQLALCHEMY_DATABASE_URL) AsyncSessionLocal sessionmaker(engine, class_AsyncSession, expire_on_commitFalse) Base declarative_base() async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close() app.post(/items/) async def create_item(item: Item, db: AsyncSession Depends(get_db)): db_item ItemModel(**item.dict()) db.add(db_item) await db.commit() await db.refresh(db_item) return db_item12. 中间件与 CORS 配置12.1 自定义中间件import 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 response app.middleware(http) async def log_requests(request: Request, call_next): print(f收到请求: {request.method} {request.url}) response await call_next(request) print(f请求处理完成: {response.status_code}) return response12.2 CORS 配置from fastapi.middleware.cors import CORSMiddleware # 添加 CORS 中间件 app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000, https://myapp.com], # 允许的源 allow_credentialsTrue, allow_methods[*], # 允许所有方法 allow_headers[*], # 允许所有头 )13. 文件上传与静态文件服务13.1 文件上传from fastapi import UploadFile, File import shutil import os UPLOAD_DIR uploads os.makedirs(UPLOAD_DIR, exist_okTrue) app.post(/upload/) async def upload_file(file: UploadFile File(...)): # 验证文件类型 if not file.filename.lower().endswith((.png, .jpg, .jpeg, .pdf)): raise HTTPException(400, 不支持的文件类型) file_path os.path.join(UPLOAD_DIR, file.filename) # 保存文件 with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) return { filename: file.filename, content_type: file.content_type, size: os.path.getsize(file_path) } app.post(/upload-multiple/) async def upload_multiple_files(files: List[UploadFile] File(...)): results [] for file in files: file_path os.path.join(UPLOAD_DIR, file.filename) with open(file_path, wb) as buffer: shutil.copyfileobj(file.file, buffer) results.append({ filename: file.filename, size: os.path.getsize(file_path) }) return {uploaded_files: results}13.2 静态文件服务from fastapi.staticfiles import StaticFiles # 挂载静态文件目录 app.mount(/static, StaticFiles(directorystatic), namestatic) app.mount(/uploads, StaticFiles(directoryuploads), nameuploads)14. 身份验证与安全性14.1 JWT 令牌认证from jose import JWTError, jwt from passlib.context import CryptContext from datetime import datetime, timedelta # 安全配置 SECRET_KEY your-secret-key ALGORITHM HS256 ACCESS_TOKEN_EXPIRE_MINUTES 30 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) 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(minutes15) to_encode.update({exp: expire}) encoded_jwt jwt.encode(to_encode, SECRET_KEY, algorithmALGORITHM) return encoded_jwt app.post(/token) async def login_for_access_token(form_data: OAuth2PasswordRequestForm Depends()): # 验证用户名密码这里简化处理 if form_data.username ! testuser or form_data.password ! testpass: raise HTTPException(401, 用户名或密码错误) access_token create_access_token(data{sub: form_data.username}) return {access_token: access_token, token_type: bearer}14.2 OAuth2 密码流from fastapi.security import OAuth2PasswordBearer oauth2_scheme OAuth2PasswordBearer(tokenUrltoken) async def get_current_user(token: str Depends(oauth2_scheme)): credentials_exception HTTPException( 401, 无法验证凭证, headers{WWW-Authenticate: Bearer}, ) try: payload jwt.decode(token, SECRET_KEY, algorithms[ALGORITHM]) username: str payload.get(sub) if username is None: raise credentials_exception except JWTError: raise credentials_exception # 这里应该从数据库获取用户信息 return {username: username} app.get(/users/me) async def read_users_me(current_user: dict Depends(get_current_user)): return current_user15. 测试与调试15.1 使用 TestClient 测试from fastapi.testclient import TestClient client TestClient(app) def test_read_main(): response client.get(/) assert response.status_code 200 assert response.json() {message: 欢迎使用 FastAPI, status: 运行正常} def test_create_item(): response client.post( /items/, json{name: 测试物品, price: 9.99} ) assert response.status_code 200 data response.json() assert data[name] 测试物品 assert data[price] 9.99 def test_read_nonexistent_item(): response client.get(/items/999) assert response.status_code 40415.2 调试技巧# 在开发时启用调试模式 if __name__ __main__: import uvicorn uvicorn.run( main:app, host0.0.0.0, port8000, reloadTrue, # 开发时启用热重载 debugTrue, # 启用调试模式 log_leveldebug ) # 添加详细的日志记录 import logging logging.basicConfig(levellogging.DEBUG)16. 部署到生产环境16.1 使用 Uvicorn 部署# 生产环境启动 uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 # 使用 Gunicorn 作为进程管理器Linux/macOS gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app16.2 Docker 部署创建DockerfileFROM python:3.9 WORKDIR /app COPY requirements.txt . RUN pip install -r requirements.txt COPY . . CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 80]创建docker-compose.ymlversion: 3.8 services: web: build: . ports: - 8000:80 environment: - DATABASE_URLpostgresql://user:passdb:5432/mydb depends_on: - db db: image: postgres:13 environment: POSTGRES_DB: mydb POSTGRES_USER: user POSTGRES_PASSWORD: pass16.3 环境变量配置from pydantic import BaseSettings class Settings(BaseSettings): database_url: str sqlite:///./test.db secret_key: str your-secret-key algorithm: str HS256 class Config: env_file .env settings Settings()17. 性能优化建议17.1 数据库优化# 使用连接池 from sqlalchemy.pool import QueuePool engine create_engine( DATABASE_URL, poolclassQueuePool, pool_size10, max_overflow20, pool_pre_pingTrue ) # 异步数据库操作 async def get_items(skip: int 0, limit: int 10): async with AsyncSessionLocal() as session: result await session.execute( select(ItemModel).offset(skip).limit(limit) ) return result.scalars().all()17.2 缓存优化from redis import asyncio as aioredis import json redis aioredis.from_url(redis://localhost) app.get(/items/{item_id}) async def read_item(item_id: int, db Depends(get_db)): # 先尝试从 Redis 缓存获取 cached await redis.get(fitem:{item_id}) if cached: return json.loads(cached) # 缓存未命中查询数据库 item db.query(ItemModel).filter(ItemModel.id item_id).first() if item is None: raise HTTPException(404, 物品未找到) # 将结果缓存 5 分钟 await redis.setex( fitem:{item_id}, 300, # 5分钟 json.dumps({id: item.id, name: item.name, price: item.price}) ) return item18. 常见问题与解决方案18.1 依赖安装问题问题安装fastapi[standard]时出现权限错误解决使用虚拟环境或添加--user参数python -m venv venv source venv/bin/activate # Linux/macOS venv\Scripts\activate # Windows pip install fastapi[standard]18.2 端口冲突问题问题端口 8000 已被占用解决更换端口或停止占用进程# 更换端口 uvicorn main:app --port 8080 # 查找占用端口的进程 netstat -ano | findstr :8000 # Windows lsof -i :8000 # Linux/macOS18.3 CORS 问题问题前端应用无法访问 API解决正确配置 CORS 中间件from fastapi.middleware.cors import CORSMiddleware app.add_middleware( CORSMiddleware, allow_origins[http://localhost:3000], # 前端开发服务器 allow_credentialsTrue, allow_methods[*], allow_headers[*], )18.4 数据库连接问题问题数据库连接失败或超时解决检查连接字符串和网络配置# 使用连接池和超时设置 engine create_engine( DATABASE_URL, pool_pre_pingTrue, pool_recycle3600, # 1小时回收连接 connect_args{connect_timeout: 10} )19. 最佳实践总结19.1 项目结构建议myproject/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 应用入口 │ ├── models.py # Pydantic 模型 │ ├── schemas.py # 数据库模型 │ ├── crud.py # 数据库操作 │ ├── dependencies.py # 依赖项 │ ├── routers/ # 路由模块 │ │ ├── __init__.py │ │ ├── items.py │ │ └── users.py │ └── config.py # 配置管理 ├── tests/ # 测试文件 ├── requirements.txt # 依赖列表 └── Dockerfile # Docker 配置19.2 代码组织原则使用路由模块化组织代码分离业务逻辑和数据模型使用依赖注入管理共享资源为生产环境配置合适的日志和监控编写完整的单元测试和集成测试19.3 安全实践永远不要将敏感信息硬编码在代码中使用环境变量管理配置实施适当的身份验证和授权验证所有输入数据限制文件上传的类型和大小FastAPI 作为一个现代、高效的 Web 框架通过其强大的类型提示支持、自动文档生成和优秀的性能表现已经成为 Python API 开发的首选框架之一。无论是小型项目还是大型企业级应用FastAPI 都能提供出色的开发体验和运行性能。