本文详解FastAPI的核心特性、项目实战和部署方案，打造现代化的Python后端服务。

前言

Python Web框架那么多，为什么选FastAPI？

• 性能最强：基于Starlette，性能媲美Node.js和Go
• 类型提示：Python类型注解，IDE智能提示
• 自动文档：Swagger/ReDoc自动生成
• 异步支持：原生async/await

今天来全面学习FastAPI。

---

一、FastAPI简介

1.1 对比其他框架

框架	性能	异步	类型提示	自动文档
FastAPI	⭐⭐⭐⭐⭐	✅	✅	✅
Flask	⭐⭐⭐	❌	❌	❌
Django	⭐⭐	⚠️	❌	❌
Tornado	⭐⭐⭐⭐	✅	❌	❌

1.2 核心特性

markdown

体验AI代码助手

代码解读

复制代码

1. 高性能 - 基于Starlette（ASGI） - 性能接近Node.js/Go 2. 开发效率 - 自动数据验证（Pydantic） - 自动API文档 - IDE智能补全 3. 标准化 - OpenAPI规范 - JSON Schema - OAuth2/JWT支持

---

二、环境搭建

2.1 安装

bash

体验AI代码助手

代码解读

复制代码

# 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装FastAPI pip install fastapi # 安装ASGI服务器 pip install "uvicorn[standard]" # 常用依赖 pip install python-multipart # 表单 pip install python-jose[cryptography] # JWT pip install passlib[bcrypt] # 密码加密 pip install sqlalchemy # ORM pip install aiosqlite # 异步SQLite

2.2 第一个应用

python

体验AI代码助手

代码解读

复制代码

# main.py from fastapi import FastAPI app = FastAPI( title="我的API", description="FastAPI学习项目", version="1.0.0" ) @app.get("/") async def root(): return {"message": "Hello FastAPI!"} @app.get("/items/{item_id}") async def read_item(item_id: int, q: str = None): return {"item_id": item_id, "q": q}

2.3 运行

bash

体验AI代码助手

代码解读

复制代码

# 开发模式（自动重载） uvicorn main:app --reload --host 0.0.0.0 --port 8000 # 访问 # API: http://localhost:8000 # Swagger文档: http://localhost:8000/docs # ReDoc文档: http://localhost:8000/redoc

---

三、核心功能

3.1 路径参数

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, Path app = FastAPI() @app.get("/users/{user_id}") async def get_user( user_id: int = Path(..., title="用户ID", ge=1) # >=1 ): return {"user_id": user_id} @app.get("/files/{file_path:path}") # 匹配路径 async def read_file(file_path: str): return {"file_path": file_path}

3.2 查询参数

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, Query from typing import Optional, List app = FastAPI() @app.get("/items/") async def list_items( skip: int = 0, limit: int = Query(default=10, le=100), # 最大100 q: Optional[str] = Query(None, min_length=3, max_length=50), tags: List[str] = Query(default=[]) ): return { "skip": skip, "limit": limit, "q": q, "tags": tags }

3.3 请求体（Pydantic模型）

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI from pydantic import BaseModel, Field, EmailStr from typing import Optional from datetime import datetime app = FastAPI() class UserCreate(BaseModel): username: str = Field(..., min_length=3, max_length=20) email: EmailStr password: str = Field(..., min_length=6) age: Optional[int] = Field(None, ge=0, le=150) class Config: json_schema_extra = { "example": { "username": "张三", "email": "zhangsan@example.com", "password": "123456", "age": 25 } } class UserResponse(BaseModel): id: int username: str email: str created_at: datetime @app.post("/users/", response_model=UserResponse) async def create_user(user: UserCreate): # 创建用户逻辑 return { "id": 1, "username": user.username, "email": user.email, "created_at": datetime.now() }

3.4 表单和文件上传

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, File, UploadFile, Form from typing import List app = FastAPI() # 表单数据 @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), "content_type": file.content_type } # 多文件上传 @app.post("/uploads/") async def upload_files(files: List[UploadFile] = File(...)): return {"filenames": [f.filename for f in files]}

---

四、依赖注入

4.1 基本依赖

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, Depends, HTTPException, status app = FastAPI() # 简单依赖 async def common_parameters(q: str = None, skip: int = 0, limit: int = 10): return {"q": q, "skip": skip, "limit": limit} @app.get("/items/") async def list_items(commons: dict = Depends(common_parameters)): return commons # 类作为依赖 class Pagination: def __init__(self, page: int = 1, size: int = 10): self.page = page self.size = size self.skip = (page - 1) * size @app.get("/users/") async def list_users(pagination: Pagination = Depends()): return { "page": pagination.page, "size": pagination.size, "skip": pagination.skip }

4.2 数据库依赖

python

体验AI代码助手

代码解读

复制代码

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.orm import sessionmaker # 数据库配置 DATABASE_URL = "sqlite+aiosqlite:///./app.db" engine = create_async_engine(DATABASE_URL, echo=True) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # 数据库依赖 async def get_db(): async with AsyncSessionLocal() as session: try: yield session finally: await session.close() # 使用 @app.get("/users/{user_id}") async def get_user(user_id: int, db: AsyncSession = Depends(get_db)): # 使用db进行数据库操作 pass

---

五、认证与授权

5.1 JWT认证

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm from jose import JWTError, jwt from passlib.context import CryptContext from datetime import datetime, timedelta from pydantic import BaseModel app = FastAPI() # 配置 SECRET_KEY = "your-secret-key-keep-it-secret" ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 30 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") # 模型 class Token(BaseModel): access_token: str token_type: str class User(BaseModel): username: str email: str disabled: bool = False # 模拟用户数据库 fake_users_db = { "admin": { "username": "admin", "email": "admin@example.com", "hashed_password": pwd_context.hash("admin123"), "disabled": False } } # 工具函数 def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def create_access_token(data: dict, expires_delta: timedelta = None): to_encode = data.copy() expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15)) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) async def get_current_user(token: str = Depends(oauth2_scheme)): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="认证失败", 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 user = fake_users_db.get(username) if user is None: raise credentials_exception return User(**user) # 登录接口 @app.post("/token", response_model=Token) async def login(form_data: OAuth2PasswordRequestForm = Depends()): user = fake_users_db.get(form_data.username) if not user or not verify_password(form_data.password, user["hashed_password"]): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="用户名或密码错误" ) access_token = create_access_token( data={"sub": user["username"]}, expires_delta=timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES) ) return {"access_token": access_token, "token_type": "bearer"} # 受保护的接口 @app.get("/users/me", response_model=User) async def read_users_me(current_user: User = Depends(get_current_user)): return current_user

---

六、中间件与异常处理

6.1 中间件

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware import time app = FastAPI() # CORS中间件 app.add_middleware( CORSMiddleware, allow_origins=["*"], # 生产环境指定域名 allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # 自定义中间件 @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 response

6.2 异常处理

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, HTTPException, Request from fastapi.responses import JSONResponse app = FastAPI() # 自定义异常 class CustomException(Exception): def __init__(self, code: int, message: str): self.code = code self.message = message # 异常处理器 @app.exception_handler(CustomException) async def custom_exception_handler(request: Request, exc: CustomException): return JSONResponse( status_code=exc.code, content={"code": exc.code, "message": exc.message} ) # 全局异常处理 @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): return JSONResponse( status_code=500, content={"code": 500, "message": "服务器内部错误"} ) # 使用 @app.get("/items/{item_id}") async def read_item(item_id: int): if item_id < 0: raise CustomException(code=400, message="ID不能为负数") return {"item_id": item_id}

---

七、项目结构

7.1 推荐结构

bash

体验AI代码助手

代码解读

复制代码

project/ ├── app/ │ ├── __init__.py │ ├── main.py # 入口 │ ├── config.py # 配置 │ ├── database.py # 数据库 │ ├── models/ # 数据模型 │ │ ├── __init__.py │ │ └── user.py │ ├── schemas/ # Pydantic模型 │ │ ├── __init__.py │ │ └── user.py │ ├── routers/ # 路由 │ │ ├── __init__.py │ │ ├── users.py │ │ └── items.py │ ├── services/ # 业务逻辑 │ │ ├── __init__.py │ │ └── user_service.py │ └── utils/ # 工具函数 │ ├── __init__.py │ └── auth.py ├── tests/ # 测试 ├── requirements.txt └── README.md

7.2 路由组织

python

体验AI代码助手

代码解读

复制代码

# app/routers/users.py from fastapi import APIRouter, Depends router = APIRouter( prefix="/users", tags=["用户管理"] ) @router.get("/") async def list_users(): return [] @router.get("/{user_id}") async def get_user(user_id: int): return {"user_id": user_id} # app/main.py from fastapi import FastAPI from app.routers import users, items app = FastAPI() app.include_router(users.router) app.include_router(items.router)

---

八、部署方案

8.1 Docker部署

dockerfile

体验AI代码助手

代码解读

复制代码

# Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

yaml

体验AI代码助手

代码解读

复制代码

# docker-compose.yml version: '3.8' services: api: build: . ports: - "8000:8000" environment: - DATABASE_URL=postgresql://user:pass@db:5432/mydb depends_on: - db restart: unless-stopped db: image: postgres:15 environment: - POSTGRES_USER=user - POSTGRES_PASSWORD=pass - POSTGRES_DB=mydb volumes: - postgres_data:/var/lib/postgresql/data volumes: postgres_data:

8.2 Nginx反向代理

nginx

体验AI代码助手

代码解读

复制代码

server { listen 80; server_name api.example.com; location / { proxy_pass http://127.0.0.1:8000; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }

8.3 远程访问

部署在内网服务器的API，如何让外网访问？

markdown

体验AI代码助手

代码解读

复制代码

方案对比： 1. 公网暴露 + HTTPS → 需要公网IP，安全风险 2. Cloudflare Tunnel → 配置复杂 3. 组网软件 → 最简单安全 使用组网软件（如星空组网）： - 服务器安装组网客户端 - 开发机安装组网客户端 - 通过虚拟IP访问API 适用场景： - 开发测试阶段 - 内部系统 - 不想暴露公网的API

---

九、性能优化

9.1 异步数据库

python

体验AI代码助手

代码解读

复制代码

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine from sqlalchemy.future import select # 异步查询 async def get_users(db: AsyncSession): result = await db.execute(select(User)) return result.scalars().all()

9.2 后台任务

python

体验AI代码助手

代码解读

复制代码

from fastapi import FastAPI, BackgroundTasks app = FastAPI() def send_email(email: str, message: str): # 发送邮件的耗时操作 pass @app.post("/notify/") async def send_notification( email: str, background_tasks: BackgroundTasks ): background_tasks.add_task(send_email, email, "通知消息") return {"message": "通知已发送"}

9.3 缓存

python

体验AI代码助手

代码解读

复制代码

from functools import lru_cache from fastapi import FastAPI app = FastAPI() @lru_cache() def get_settings(): # 读取配置（只执行一次） return {"app_name": "MyAPI"} @app.get("/settings/") async def read_settings(): return get_settings()

---

十、总结

FastAPI核心要点：

功能	方法
路径参数	@app.get("/items/{id}")
查询参数	def read(q: str = None)
请求体	def create(item: Item)
依赖注入	Depends()
认证	OAuth2PasswordBearer
中间件	@app.middleware("http")

FastAPI优势：

• Python最快的Web框架
• 自动生成API文档
• 类型安全，IDE友好
• 异步原生支持