简介:本文详细介绍FastAPI框架的核心特性、安装步骤、基础用法及进阶技巧,通过代码示例与场景分析帮助开发者快速掌握现代Web API开发。
FastAPI是2018年推出的现代化Python Web框架,基于Starlette(ASGI框架)和Pydantic(数据验证库)构建,专为高性能API开发设计。其核心优势体现在三方面:
典型应用场景包括:微服务架构、机器学习模型服务、实时数据接口、高并发Web服务。GitHub数据显示,截至2023年Q3,FastAPI在Python Web框架中的周下载量已突破800万次。
# 创建虚拟环境(推荐)python -m venv fastapi_envsource fastapi_env/bin/activate # Linux/macOSfastapi_env\Scripts\activate # Windows# 核心库安装pip install fastapi uvicorn[standard]# 可选扩展pip install python-multipart # 表单支持pip install sqlalchemy # ORM集成pip install aiohttp # 异步HTTP客户端
from fastapi import FastAPIapp = FastAPI()@app.get("/")async def read_root():return {"message": "Welcome to FastAPI"}@app.get("/items/{item_id}")async def read_item(item_id: int, q: str = None):return {"item_id": item_id, "q": q}
运行命令:
uvicorn main:app --reload --host 0.0.0.0 --port 8000
访问http://localhost:8000/docs可查看自动生成的Swagger UI文档。
from fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):name: strdescription: str | None = Noneprice: floattax: float | None = None@app.post("/items/")async def create_item(item: Item):item_dict = item.dict()if item.tax:price_with_tax = item.price + item.taxitem_dict.update({"price_with_tax": price_with_tax})return item_dict
关键特性:
| 装饰器 | HTTP方法 | 典型场景 |
|---|---|---|
@app.get |
GET | 数据查询 |
@app.post |
POST | 资源创建 |
@app.put |
PUT | 资源替换 |
@app.patch |
PATCH | 资源部分更新 |
@app.delete |
DELETE | 资源删除 |
from fastapi import Depends, FastAPI, HTTPExceptionapp = FastAPI()async def verify_token(x_token: str = Header(...)):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")@app.get("/items/", dependencies=[Depends(verify_token)])async def read_items():return [{"item": "Foo"}, {"item": "Bar"}]
依赖项可复用场景:
from fastapi import FastAPIimport httpxapp = FastAPI()async def fetch_data(url: str):async with httpx.AsyncClient() as client:return await client.get(url)@app.get("/proxy/")async def proxy_request(url: str):response = await fetch_data(url)return response.json()
异步优势:
from fastapi import FastAPI, WebSocketfrom fastapi.responses import HTMLResponseapp = FastAPI()html = """<html><body><h1>WebSocket Test</h1><form action="" onsubmit="sendMessage(event)"><input type="text" id="messageText" autocomplete="off"/><button>Send</button></form><ul id='messages'></ul><script>const ws = new WebSocket("ws://localhost:8000/ws");ws.onmessage = function(event) {const messages = document.getElementById('messages')const messageItem = document.createElement('li')messageItem.textContent = event.datamessages.appendChild(messageItem)};function sendMessage(event) {const input = document.getElementById("messageText")ws.send(input.value)input.value = ''event.preventDefault()}</script></body></html>"""@app.get("/")async def get():return HTMLResponse(html)@app.websocket("/ws")async def websocket_endpoint(websocket: WebSocket):await websocket.accept()while True:data = await websocket.receive_text()await websocket.send_text(f"Message text was: {data}")
from fastapi import FastAPI, Requestimport timeapp = FastAPI()@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_timeresponse.headers["X-Process-Time"] = str(process_time)return response
中间件典型用途:
# uvicorn启动参数建议uvicorn main:app --host 0.0.0.0 --port 8000 \--workers 4 \ # worker数量=2*CPU核心数+1--timeout-keep-alive 60 \--limit-concurrency 100 \--backlog 2048
缓存策略:
cachetools实现内存缓存数据库优化:
from sqlalchemy.ext.asyncio import AsyncSessionfrom contextlib import asynccontextmanager@asynccontextmanagerasync def async_session():async with async_engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)async with AsyncSession(bind=async_engine) as session:yield session
请求负载控制:
from fastapi import FastAPI, Request, Responsefrom fastapi.middleware import Middlewarefrom fastapi.middleware.base import BaseHTTPMiddlewareclass RateLimitMiddleware(BaseHTTPMiddleware):def __init__(self, app, requests_per_minute=60):super().__init__(app)self.requests_per_minute = requests_per_minute# 实现令牌桶算法等限流逻辑app = FastAPI()app.add_middleware(RateLimitMiddleware, requests_per_minute=120)
CORS配置:
from fastapi.middleware.cors import CORSMiddlewareapp.add_middleware(CORSMiddleware,allow_origins=["*"], # 生产环境应明确指定allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)
敏感数据保护:
python-dotenv管理环境变量
from fastapi.testclient import TestClientfrom main import appclient = TestClient(app)def test_read_main():response = client.get("/")assert response.status_code == 200assert response.json() == {"message": "Welcome to FastAPI"}def test_create_item():item_data = {"name": "Test Item","price": 100.0}response = client.post("/items/", json=item_data)assert response.status_code == 200assert response.json()["name"] == "Test Item"
日志配置:
import logginglogging.basicConfig(level=logging.DEBUG,format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",handlers=[logging.FileHandler("app.log"),logging.StreamHandler()])
异常处理:
from fastapi import FastAPI, HTTPExceptionapp = FastAPI()@app.exception_handler(HTTPException)async def http_exception_handler(request, exc):return JSONResponse(status_code=exc.status_code,content={"message": exc.detail, "error": str(exc)})
| 部署方式 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| Uvicorn直接运行 | 开发/测试环境 | 配置简单 | 缺乏进程管理 |
| Gunicorn+Uvicorn | 生产环境(中小规模) | 进程管理、负载均衡 | 配置复杂度中等 |
| Docker容器化 | 云原生部署 | 环境一致性、可扩展性 | 需要掌握Docker技术 |
| Kubernetes集群 | 高可用、大规模部署 | 自动扩缩容、服务发现 | 运维复杂度高 |
典型Dockerfile示例:
FROM python:3.9-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
| 库名 | 功能 | 版本要求 |
|---|---|---|
| fastapi-users | 完整认证系统 | 10.x |
| sqlmodel | SQLAlchemy+Pydantic集成 | 1.x |
| fastapi-cache | 多级缓存支持 | 0.8.x |
| httpx | 异步HTTP客户端 | 0.23.x |
from fastapi import FastAPIfrom starlette.middleware.base import BaseHTTPMiddlewarefrom starlette.requests import Requestfrom starlette.responses import Responseclass CustomPlugin(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):# 前置处理逻辑response = await call_next(request)# 后置处理逻辑return responseapp = FastAPI()app.add_middleware(CustomPlugin)
CORS错误:
allow_origins类型验证失败:
from fastapi import HTTPExceptionfrom pydantic import ValidationError@app.exception_handler(ValidationError)async def validation_exception_handler(request, exc):return JSONResponse(status_code=422,content={"detail": [e.__str__() for e in exc.errors()]},)
异步超时处理:
from fastapi import FastAPIfrom starlette.concurrency import run_in_threadpoolimport asyncioapp = FastAPI()async def blocking_task():await asyncio.sleep(5) # 模拟阻塞操作@app.get("/")async def read_root():try:await asyncio.wait_for(blocking_task(), timeout=2.0)except asyncio.TimeoutError:raise HTTPException(status_code=408, detail="Request timeout")return {"message": "Success"}
通过系统学习本文内容,开发者可掌握FastAPI的核心开发技能,从基础路由构建到生产环境部署形成完整知识体系。建议结合官方文档与实战项目进行深入练习,逐步积累现代Web API开发经验。