FastAPI, 一個神奇的 Python 庫
作者:老朱2000
FastAPI 是一個用于構建 API 的現代、快速(高性能)的 web 框架,使用 Python 并基于標準的 Python類型提示。
FastAPI 是一個用于構建 API 的現代、快速(高性能)的 web 框架,使用 Python 并基于標準的 Python類型提示。
安裝
安裝 FastAPI 很簡單,這里我們使用 pip 命令來安裝。
pip install fastapi
另外我們還需要一個 ASGI 服務器,生產環境可以使用 Uvicorn 或者 Hypercorn
pip install uvicorn[standard]
資料:
- FastAPI 文檔:https://fastapi.tiangolo.com/zh/
- FastAPI 源碼:https://github.com/tiangolo/fastapi
簡單示例
# main.py
from fastapi import FastAPI
app = FastAPI()
@app.get("/")
def read_root():
return {"Hello": "World"}
啟動服務:
uvicorn main:app --reload
- --reload:開發時自動重載(生產環境移除)
- 訪問:http://localhost:8000
關鍵功能
(1) 路徑參數和查詢參數
- {user_id}:路徑參數(必須)
- limit 和 skip:查詢參數(可選,帶默認值)
@app.get("/users/{user_id}")
async def read_user(user_id: int,
limit: int = 10,
skip: int = 0):
return {"user_id": user_id, "limit": limit, "skip": skip}
(2) 請求體(POST/PUT)
使用 Pydantic 模型驗證數據:
from pydantic import BaseModel
class Item(BaseModel):
name: str
description: str = None
price: float
tax: float = None
@app.post("/items/")
async def create_item(item: Item):
return {"item": item.dict()}
(3) 自動文檔
- Swagger UI:http://localhost:8000/docs
- ReDoc:http://localhost:8000/redoc
(4) 異步支持
@app.get("/slow-endpoint")
async def slow_operation():
# 如數據庫查詢
await some_async_task()
return {"status": "done"}
進階功能
(1) 依賴注入
復用代碼邏輯(如認證、數據庫連接)。
from fastapi import Depends
def common_params(q: str = None, skip: int = 0, limit: int = 100):
return {"q": q, "skip": skip, "limit": limit}
@app.get("/items/")
async def read_items(commons: dict = Depends(common_params)):
return commons
(2) 中間件
處理請求前/后的邏輯(如 CORS、日志)。
from fastapi.middleware.cors import CORSMiddleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
)
(3) WebSocket 支持
@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: {data}")
(4) 后臺任務
執行無需即時響應的操作(如發送郵件、短信)
from fastapi import BackgroundTasks
def log_task(message: str):
with open("log.txt", "a") as f:
f.write(message)
@app.post("/send-email")
async def send_email(background_tasks: BackgroundTasks):
background_tasks.add_task(log_task, "Email sent")
return {"status": "ok"}
適用場景
- 構建高性能 REST API
- 微服務后端
- 實時應用(WebSocket)
- 結合sqlalchemy進行 web 開發
責任編輯:趙寧寧
來源:
程序員老朱