简介:本文通过系统化的FastAPI入门教程,结合代码示例与开发实践,帮助开发者快速掌握现代Web API开发技能,涵盖环境搭建、核心功能实现及生产部署全流程。
FastAPI作为基于Python的现代Web框架,凭借其高性能、自动文档生成和异步支持特性,已成为构建API服务的首选工具。本文通过环境配置、基础路由、数据验证、异步处理及部署优化等模块,系统讲解FastAPI核心开发流程,提供可复用的代码模板与最佳实践,帮助开发者在2小时内完成从安装到生产部署的全流程学习。
FastAPI基于Starlette(异步框架)和Pydantic(数据验证),在TechEmpower基准测试中性能接近Go/Node.js水平。其核心设计采用ASGI标准,支持异步请求处理,单核QPS可达传统WSGI框架的3-5倍。
完美支持异步数据库驱动(如asyncpg)、消息队列(Redis/RabbitMQ)及微服务架构,与Docker/Kubernetes无缝集成。
# 创建虚拟环境(推荐Python 3.8+)python -m venv fastapi_envsource fastapi_env/bin/activate # Linux/Mac# Windows: .\fastapi_env\Scripts\activate# 安装核心依赖pip install fastapi uvicorn[standard]
--reload参数实现热重载
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}
关键点:
from fastapi import FastAPI, Queryfrom 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
验证机制:
@validator装饰器)
from fastapi import Dependsimport databasesimport sqlalchemyDATABASE_URL = "sqlite:///./test.db"database = databases.Database(DATABASE_URL)metadata = sqlalchemy.MetaData()notes = sqlalchemy.Table("notes",metadata,sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),sqlalchemy.Column("text", sqlalchemy.String),sqlalchemy.Column("completed", sqlalchemy.Boolean),)engine = sqlalchemy.create_engine(DATABASE_URL)metadata.create_all(engine)@app.on_event("startup")async def startup():await database.connect()@app.on_event("shutdown")async def shutdown():await database.disconnect()@app.get("/notes/")async def read_notes():query = notes.select()return await database.fetch_all(query)
最佳实践:
on_event管理数据库连接生命周期database.transaction()实现
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4 --timeout 120
参数说明:
workers:根据CPU核心数设置(通常为2n+1)timeout:处理长请求的超时设置--proxy-headers:在反向代理后使用时启用
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"]
构建命令:
docker build -t fastapi-app .docker run -d -p 8000:8000 fastapi-app
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;}}
from fastapi import FastAPI, Requestfrom fastapi.responses import JSONResponsefrom cachetools import TTLCacheapp = FastAPI()cache = TTLCache(maxsize=100, ttl=300) # 5分钟缓存@app.get("/expensive-operation/{item_id}")async def get_expensive_data(item_id: int, request: Request):cache_key = f"item_{item_id}"if cache_key in cache:return cache[cache_key]# 模拟耗时操作result = {"data": f"Processed {item_id}", "timestamp": "now"}cache[cache_key] = resultreturn result
from fastapi import FastAPI, Requestimport timeapp = FastAPI()@app.middleware("http")async def add_timing_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
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],)
from fastapi.staticfiles import StaticFilesapp.mount("/static", StaticFiles(directory="static"), name="static")
通过系统掌握上述内容,开发者可在3天内完成从FastAPI入门到生产环境部署的全流程。建议结合官方文档(fastapi.tiangolo.com)进行深度学习,并参与GitHub开源项目实践提升实战能力。