简介:本文通过FastAPI框架,从环境搭建到API开发、性能优化,提供一套完整的Web API项目开发指南,助力开发者快速构建高效服务。
在Python生态中,Flask和Django长期占据Web开发的主导地位,但随着微服务架构和API经济的兴起,开发者对高性能、低延迟、强类型支持的框架需求日益迫切。FastAPI作为后起之秀,凭借其基于标准Python类型注解的自动API文档生成、原生异步支持和接近原生ASGI的极致性能,迅速成为开发Web API的首选框架。本文将通过一个完整的项目案例,详细解析如何利用FastAPI快速构建一个高效的Web API服务。
FastAPI基于Python 3.7+,需安装以下核心依赖:
pip install fastapi uvicorn
合理的项目结构能提升代码可维护性。推荐采用以下分层架构:
project/├── app/ # 主应用目录│ ├── __init__.py # 初始化模块│ ├── main.py # 入口文件│ ├── routers/ # 路由模块│ │ ├── __init__.py│ │ └── items.py # 示例路由│ ├── models/ # 数据模型│ │ └── item.py # Pydantic模型│ └── dependencies/ # 依赖注入│ └── auth.py # 认证依赖└── requirements.txt # 依赖列表
在main.py中编写第一个FastAPI应用:
from fastapi import FastAPIapp = FastAPI()@app.get("/")async def read_root():return {"message": "Hello FastAPI!"}
运行命令:
uvicorn app.main:app --reload
访问http://127.0.0.1:8000,即可看到返回的JSON响应。--reload参数启用开发模式下的代码热重载。
FastAPI通过装饰器定义路由,支持所有HTTP方法:
from fastapi import FastAPI, HTTPExceptionfrom typing import Optionalapp = FastAPI()# 模拟数据库fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}]@app.get("/items/{item_id}")async def read_item(item_id: int, q: Optional[str] = None):if q:return {"item_id": item_id, "q": q}return {"item_id": item_id}@app.post("/items/")async def create_item(item: dict):fake_items_db.append(item)return {"message": "Item created"}
{item_id}通过类型注解自动转换为整数。q: Optional[str]表示可选的查询字符串参数。item: dict接收JSON数据。使用Pydantic定义严格的数据模型:
# models/item.pyfrom pydantic import BaseModelclass Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = None
在路由中使用模型:
from app.models.item import Item@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
Item模型定义。item.dict()将模型转换为字典。FastAPI的依赖注入系统支持复用逻辑(如数据库连接、认证):
# dependencies/auth.pyfrom fastapi import Depends, Header, HTTPExceptionasync def get_token_header(x_token: str = Header(...)):if x_token != "fake-super-secret-token":raise HTTPException(status_code=400, detail="X-Token header invalid")return x_token@app.get("/items/")async def read_items(token: str = Depends(get_token_header)):return [{"item_name": "Foo"}, {"item_name": "Bar"}]
Header(...)强制要求X-Token头。get_token_header可在多个路由中复用。FastAPI原生支持async/await,适合I/O密集型操作:
import httpxasync def fetch_data(url: str):async with httpx.AsyncClient() as client:return await client.get(url)@app.get("/external-data/")async def get_external_data():response = await fetch_data("https://example.com")return response.json()
使用pytest和httpx编写测试:
# tests/test_main.pyfrom fastapi.testclient import TestClientfrom app.main import appclient = TestClient(app)def test_read_root():response = client.get("/")assert response.status_code == 200assert response.json() == {"message": "Hello FastAPI!"}def test_create_item():response = client.post("/items/",json={"name": "Test Item", "price": 10.5},)assert response.status_code == 200assert "Test Item" in response.json()["name"]
运行测试:
pytest tests/
生产环境需禁用--reload,并指定主机和端口:
uvicorn app.main:app --host 0.0.0.0 --port 8000 --workers 4
--workers根据CPU核心数设置(通常为2*CPU+1)。配置Nginx作为反向代理:
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;}}
创建Dockerfile:
FROM python:3.9-slimWORKDIR /appCOPY requirements.txt .RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
构建并运行:
docker build -t fastapi-app .docker run -d -p 8000:8000 fastapi-app
使用locust进行压力测试:
# locustfile.pyfrom locust import HttpUser, taskclass FastAPIUser(HttpUser):@taskdef get_root(self):self.client.get("/")
运行命令:
locust -f locustfile.py
对不常变的数据使用缓存:
from fastapi import FastAPI, Responsefrom cachetools import TTLCachecache = TTLCache(maxsize=100, ttl=300) # 5分钟缓存@app.get("/cached-data/")async def get_cached_data():if "data" in cache:return cache["data"]data = {"expensive": "computation"}cache["data"] = datareturn data
启用CORS:在main.py中添加:
from fastapi.middleware.cors import CORSMiddlewareapp.add_middleware(CORSMiddleware,allow_origins=["*"],allow_methods=["*"],allow_headers=["*"],)
速率限制:使用slowapi库:
FastAPI通过其类型安全、异步原生、自动文档等特性,显著降低了Web API的开发门槛。本文通过一个完整的项目案例,覆盖了从环境搭建到生产部署的全流程。实际开发中,还可结合以下扩展:
FastAPI.WebSocket实现实时通信。strawberry-graphql。FastAPI的极简设计和强大功能,使其成为现代Web API开发的理想选择。无论是个人项目还是企业级应用,都能通过FastAPI快速构建高效、可靠的服务。