Python FastAPI框架从入门到实战
FastAPI是Python生态中增长最快的Web框架,它基于Python类型提示构建,提供了自动API文档生成、请求验证、异步支持等现代特性。FastAPI的性能与Node.js和Go相当,同时保持了Python的简洁优雅。本文将带你从零开始学习FastAPI,构建一个完整的RESTful API项目。
一、快速上手
pip install fastapi uvicorn
from fastapi import FastAPI
from pydantic import BaseModel
from typing import Optional
app = FastAPI(title="用户管理API", version="1.0.0")
class UserCreate(BaseModel):
name: str
email: str
age: Optional[int] = None
class UserResponse(BaseModel):
id: int
name: str
email: str
age: Optional[int] = None
@app.post("/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
db_user = await save_to_db(user)
return db_user
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int):
user = await get_from_db(user_id)
if not user:
raise HTTPException(status_code=404, detail="用户不存在")
return user
@app.get("/users", response_model=list[UserResponse])
async def list_users(skip: int = 0, limit: int = 20, role: Optional[str] = None):
return await get_users(skip=skip, limit=limit, role=role)
二、依赖注入系统
FastAPI的依赖注入是其最强大的特性之一,可以优雅地处理认证、数据库连接等共享资源:
from fastapi import Depends, HTTPException, Header
async def get_db():
db = Database()
try:
yield db
finally:
await db.close()
async def get_current_user(
authorization: str = Header(...),
db: Database = Depends(get_db)
):
token = authorization.replace("Bearer ", "")
user = await verify_token(db, token)
if not user:
raise HTTPException(status_code=401, detail="无效的认证凭据")
return user
async def require_admin(user = Depends(get_current_user)):
if user.role != "admin":
raise HTTPException(status_code=403, detail="需要管理员权限")
return user
@app.delete("/users/{user_id}")
async def delete_user(
user_id: int,
admin = Depends(require_admin),
db = Depends(get_db)
):
await db.execute("DELETE FROM users WHERE id = :id", {"id": user_id})
return {"message": "用户已删除"}
三、自动API文档
FastAPI最大的亮点是自动生成交互式API文档。访问/docs可以看到Swagger UI,/redoc可以看到ReDoc格式的文档。你只需要定义好Pydantic模型和类型注解,文档就自动生成了,包括请求参数、响应格式、示例数据等。
四、中间件与CORS
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.com"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.middleware("http")
async def log_requests(request, call_next):
start_time = time.time()
response = await call_next(request)
duration = time.time() - start_time
logger.info(f"{request.method} {request.url} - {duration:.3f}s")
return response
五、数据库集成与异步ORM
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker, declarative_base
engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db")
async_session = sessionmaker(engine, class_=AsyncSession)
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
name = Column(String(50))
email = Column(String(100), unique=True)
# CRUD操作
async def get_users(db: AsyncSession, skip: int = 0, limit: int = 20):
result = await db.execute(select(User).offset(skip).limit(limit))
return result.scalars().all()
FastAPI结合了Python的类型系统和现代Web框架的最佳实践,让API开发变得高效而愉快。自动生成的Swagger文档、强大的依赖注入系统和原生异步支持,使它成为构建RESTful API的理想选择。建议在新项目中优先考虑FastAPI。