当前位置：首页 > news >正文

深入浅出FastAPI：现代Python Web开发的利器

news 2025/10/16 11:25:29

诸神缄默不语-个人技术博文与视频目录

还在为Python Web开发的速度和性能发愁？FastAPI或许是你一直在寻找的解决方案！

什么是FastAPI？

FastAPI是一个现代、快速（高性能）的Python Web框架，专门用于构建API。它基于标准Python类型提示，使用Python 3.6+版本的类型提示特性，让API开发变得简单而强大。

FastAPI的核心优势

⚡ 极快的性能：与NodeJS和Go相当，是Python领域最快的框架之一
🚀 高效的编码：开发速度提升约200%-300%
📝 更少的bug：减少约40%的人为错误
💡 直观的API：强大的编辑器支持，自动补全无处不在
📚 完整的标准：基于并完全兼容开放的API标准（OpenAPI和JSON Schema）

安装FastAPI

pip install fastapi
pip install uvicorn[standard]

第一个FastAPI应用

让我们从一个简单的"Hello World"开始：

from fastapi import FastAPI# 创建FastAPI实例
app = FastAPI()# 定义根路径的路由
@app.get("/")
async def read_root():return {"message": "Hello World"}# 带路径参数的路由
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):return {"item_id": item_id, "q": q}

保存为 try1.py，然后运行：

python -m uvicorn try1:app --reload

访问 http://127.0.0.1:8000，你将看到JSON响应：{"message": "Hello World"}

uvicorn也可以使用Python脚本运行：
创建一个 try2.py 文件：

import uvicornif __name__ == "__main__":uvicorn.run("try1:app", host="127.0.0.1", port=8000, reload=True)

然后运行：

python try2.py

自动API文档

FastAPI最酷的特性之一就是自动生成交互式API文档！

Swagger UI: http://127.0.0.1:8000/docs
ReDoc: http://127.0.0.1:8000/redoc

这些文档是完全交互的，你可以直接在浏览器中测试API！

由于网络问题，可能打开后无法显示内容，可以在F12中看到需要从外网下js：
在这里插入图片描述

深入FastAPI特性

1. 类型提示和数据验证

from typing import Optional
from pydantic import BaseModel# 定义数据模型
class Item(BaseModel):name: strdescription: Optional[str] = Noneprice: floattax: Optional[float] = 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

2. 路径参数和查询参数

from enum import Enumclass ModelName(str, Enum):alexnet = "alexnet"resnet = "resnet"lenet = "lenet"@app.get("/models/{model_name}")
async def get_model(model_name: ModelName):if model_name == ModelName.alexnet:return {"model_name": model_name, "message": "Deep Learning FTW!"}if model_name.value == "lenet":return {"model_name": model_name, "message": "LeCNN all the images"}return {"model_name": model_name, "message": "Have some residuals"}@app.get("/users/")
async def read_users(skip: int = 0, limit: int = 10, q: Optional[str] = None):# 模拟从数据库获取用户users = [{"id": i, "name": f"User {i}"} for i in range(100)]if q:users = [user for user in users if q.lower() in user["name"].lower()]return users[skip: skip + limit]

3. 请求体和复杂数据模型

from datetime import datetime
from typing import Listclass User(BaseModel):id: intname: str = "John Doe"signup_ts: Optional[datetime] = Nonefriends: List[int] = []class ComplexData(BaseModel):user: Useritems: List[Item]@app.put("/complex-data/{data_id}")
async def update_complex_data(data_id: int, data: ComplexData):return {"data_id": data_id,"data": data.dict(),"received_at": datetime.now()}

4. 依赖注入系统

from fastapi import Depends, HTTPException, status# 简单的依赖
async def common_parameters(skip: int = 0, limit: int = 100,q: Optional[str] = None
):return {"skip": skip, "limit": limit, "q": q}@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):return commons# 更复杂的依赖注入示例
class Database:async def get_user(self, user_id: int):# 模拟数据库查询return {"user_id": user_id, "name": f"User {user_id}"}async def get_database():return Database()async def get_current_user(user_id: int, db: Database = Depends(get_database)
):user = await db.get_user(user_id)if not user:raise HTTPException(status_code=404, detail="User not found")return user@app.get("/users/me")
async def read_current_user(current_user: dict = Depends(get_current_user)):return current_user

5. 错误处理

from fastapi import HTTPException
from fastapi.responses import JSONResponse# 自定义异常处理器
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc):return JSONResponse(status_code=exc.status_code,content={"message": exc.detail, "error": True})# 抛出HTTP异常
@app.get("/protected-route")
async def protected_route(token: str = None):if not token or token != "secret-token":raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED,detail="Invalid authentication credentials",headers={"WWW-Authenticate": "Bearer"},)return {"message": "Access granted"}

6. 中间件和CORS

from fastapi.middleware.cors import CORSMiddleware
import time# 添加CORS中间件
app.add_middleware(CORSMiddleware,allow_origins=["*"],  # 生产环境中应该指定具体域名allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 自定义中间件：添加处理时间头
@app.middleware("http")
async def add_process_time_header(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

完整示例：待办事项API

让我们构建一个完整的待办事项管理API：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List, Optional
from uuid import UUID, uuid4app = FastAPI(title="Todo API", version="1.0.0")# 数据模型
class TodoItem(BaseModel):id: Optional[UUID] = Nonetitle: strdescription: Optional[str] = Nonecompleted: bool = False# 模拟数据库
class TodoDB:def __init__(self):self.todos: List[TodoItem] = []def get_all(self) -> List[TodoItem]:return self.todosdef get(self, todo_id: UUID) -> Optional[TodoItem]:for todo in self.todos:if todo.id == todo_id:return todoreturn Nonedef create(self, todo: TodoItem) -> TodoItem:todo.id = uuid4()self.todos.append(todo)return tododef update(self, todo_id: UUID, updated_todo: TodoItem) -> Optional[TodoItem]:for idx, todo in enumerate(self.todos):if todo.id == todo_id:updated_todo.id = todo_idself.todos[idx] = updated_todoreturn updated_todoreturn Nonedef delete(self, todo_id: UUID) -> bool:for idx, todo in enumerate(self.todos):if todo.id == todo_id:del self.todos[idx]return Truereturn False# 依赖注入
def get_db():return TodoDB()# API路由
@app.get("/todos", response_model=List[TodoItem])
async def list_todos(db: TodoDB = Depends(get_db)):return db.get_all()@app.get("/todos/{todo_id}", response_model=TodoItem)
async def get_todo(todo_id: UUID, db: TodoDB = Depends(get_db)):todo = db.get(todo_id)if not todo:raise HTTPException(status_code=404, detail="Todo not found")return todo@app.post("/todos", response_model=TodoItem)
async def create_todo(todo: TodoItem, db: TodoDB = Depends(get_db)):return db.create(todo)@app.put("/todos/{todo_id}", response_model=TodoItem)
async def update_todo(todo_id: UUID, updated_todo: TodoItem, db: TodoDB = Depends(get_db)):todo = db.update(todo_id, updated_todo)if not todo:raise HTTPException(status_code=404, detail="Todo not found")return todo@app.delete("/todos/{todo_id}")
async def delete_todo(todo_id: UUID, db: TodoDB = Depends(get_db)):if not db.delete(todo_id):raise HTTPException(status_code=404, detail="Todo not found")return {"message": "Todo deleted successfully"}

部署FastAPI应用

使用Uvicorn部署：

# 开发环境（带热重载）
uvicorn main:app --reload --host 0.0.0.0 --port 8000# 生产环境
uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4

使用Docker + uvicorn部署：

FROM python:3.9WORKDIR /codeCOPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txtCOPY . .CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "80"]

查看全文

http://www.dtcms.com/a/487548.html