FastAPI 框架以其非同步特性及 Pydantic 模型的整合,大幅提升了 Web 應用程式的開發效率和效能。除了基礎的 HTTP 路由和資料處理,FastAPI 也支援 WebSockets,方便開發者建構即時通訊應用。其內建的自動檔案生成功能,更簡化了 API 的開發和維護流程。此外,FastAPI 也提供完善的安全性機制,方便整合 OAuth2、JWT 等認證方式,確保應用程式的安全性。
FastAPI 深度解析:從基礎到進階應用
FastAPI 是一個現代化的 Python Web 框架,以其高效能、易用性和豐富的功能特性,迅速成為開發者的首選。本文將探討 FastAPI 的核心概念、關鍵技術及其在實際專案中的應用實踐。
1. FastAPI 基礎架構與核心概念
FastAPI 根據 Python 3.7+ 的型別提示(Type Hints)機制,提供強大的資料驗證和自動檔案生成能力。其核心架構包括:
1.1 路徑操作函式(Path Operation Function)
路徑操作函式是 FastAPI 處理 HTTP 請求的核心元件。透過 @app.get()
、@app.post()
等裝飾器,將函式與特定的 URL 路徑和 HTTP 方法繫結。
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
class Item(BaseModel):
name: str
price: float
@app.post("/items/")
async def create_item(item: Item):
return item
內容解密:
- 使用
FastAPI()
建立應使用案例項。 - 定義 Pydantic 模型
Item
,包含name
和price
欄位。 @app.post("/items/")
裝飾器將create_item
函式與 POST 請求繫結。- 函式接收
Item
型別的引數,並自動進行資料驗證。 - 傳回處理後的
item
物件。
1.2 路徑引數與查詢引數
FastAPI 支援靈活的引數處理機制,包括路徑引數和查詢引數。
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
return {"item_id": item_id, "q": q}
內容解密:
item_id
為路徑引數,自動從 URL 中擷取並轉換為整數型別。q
為查詢引數,可選,預設值為None
。- 函式傳回包含
item_id
和q
的 JSON 物件。
2. 資料驗證與 Pydantic 模型
Pydantic 是 FastAPI 的核心依賴之一,提供強大的資料驗證功能。
2.1 基本模型定義
from pydantic import BaseModel, validator
class User(BaseModel):
username: str
email: str
@validator('email')
def email_must_be_valid(cls, v):
if '@' not in v:
raise ValueError('Invalid email')
return v
內容解密:
- 定義
User
模型,包含username
和email
欄位。 - 使用
@validator
裝飾器對email
欄位進行自定義驗證。 - 當
email
不包含@
時,引發驗證錯誤。
3. 資料函式庫整合與 ORM
FastAPI 可與多種資料函式庫和 ORM(Object-Relational Mapping)工具整合,如 SQLAlchemy。
3.1 SQLAlchemy 非同步操作
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite+aiosqlite:///./test.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL)
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async def get_db():
async with async_session() as session:
yield session
@app.post("/users/")
async def create_user(user: User, db: AsyncSession = Depends(get_db)):
db.add(user)
await db.commit()
return user
內容解密:
- 建立非同步資料函式庫引擎和會話管理器。
- 定義
get_db
依賴函式,提供資料函式庫會話。 - 在
create_user
路徑操作中使用AsyncSession
處理資料函式庫操作。
4. API 檔案與測試
FastAPI 自動生成互動式 API 檔案,根據 Swagger UI 和 ReDoc。
4.1 自定義檔案資訊
app = FastAPI(
title="My FastAPI Application",
description="This is a description of my API",
version="1.0.0",
)
內容解密:
- 在建立 FastAPI 應用時提供元資料資訊。
- 自動生成的 API 檔案包含標題、描述和版本資訊。
5. 安全性與認證授權
FastAPI 提供多種安全性功能,包括 OAuth2、JWT 認證等。
5.1 OAuth2 認證範例
from fastapi.security import OAuth2PasswordBearer
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
@app.get("/protected/")
async def protected_route(token: str = Depends(oauth2_scheme)):
return {"token": token}
內容解密:
- 建立 OAuth2 認證方案,指定 token 取得 URL。
- 在受保護的路由中使用
Depends(oauth2_scheme)
取得 token。
FastAPI 技術深度解析與應用實踐
FastAPI 是一個現代化的 Python Web 框架,以其高效、易用和強大的功能而聞名。本文將探討 FastAPI 的核心技術和應用實踐,涵蓋其主要特性、WebSockets、測試和佈署等方面。
FastAPI 核心特性與優勢
FastAPI 根據 Python 3.7+ 的型別提示(Type Hints)機制,提供了自動化的 API 檔案生成、強大的資料驗證和高效的非同步處理能力。這些特性使得開發者能夠快速構建高品質的 Web 應用程式。
自動化 API 檔案生成
FastAPI 自動生成互動式 API 檔案,方便開發者測試和除錯 API。這些檔案根據 OpenAPI 標準,支援多種格式,如 JSON 和 YAML。
from fastapi import FastAPI
app = FastAPI()
@app.get("/items/{item_id}")
async def read_item(item_id: int):
return {"item_id": item_id}
內容解密:
上述程式碼定義了一個簡單的 GET 請求處理函式 read_item
,使用型別提示指定 item_id
為整數。FastAPI 自動為此端點生成 API 檔案,並提供強大的資料驗證功能。
WebSockets 支援與實作
WebSockets 提供雙向通訊通道,允許伺服器主動向客戶端推播資料。FastAPI 原生支援 WebSockets,使得構建即時應用變得更加容易。
WebSocket 連線建立與處理
from fastapi import FastAPI, WebSocket
app = FastAPI()
@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 text was: {data}")
內容解密:
此範例展示瞭如何使用 FastAPI 建立 WebSocket 連線。websocket_endpoint
函式接受 WebSocket 連線請求,接收客戶端傳送的訊息並回傳處理結果。
多客戶端聊天應用實作
利用 FastAPI 的 WebSocket 功能,可以輕鬆構建多客戶端聊天應用。
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
app = FastAPI()
class ConnectionManager:
def __init__(self):
self.active_connections: List[WebSocket] = []
async def connect(self, websocket: WebSocket):
await websocket.accept()
self.active_connections.append(websocket)
def disconnect(self, websocket: WebSocket):
self.active_connections.remove(websocket)
async def send_personal_message(self, message: str, websocket: WebSocket):
await websocket.send_text(message)
async def broadcast(self, message: str):
for connection in self.active_connections:
await connection.send_text(message)
manager = ConnectionManager()
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
await manager.connect(websocket)
try:
while True:
data = await websocket.receive_text()
await manager.broadcast(f"Client {client_id} says: {data}")
except WebSocketDisconnect:
manager.disconnect(websocket)
內容解密:
此範例實作了一個簡單的多客戶端聊天系統。ConnectionManager
類別負責管理所有活躍的 WebSocket 連線,並提供廣播訊息的功能。每當客戶端傳送訊息時,伺服器將訊息廣播給所有連線的客戶端。
測試與佈署
FastAPI 提供了強大的測試功能,支援使用 TestClient
對應用進行單元測試和整合測試。
使用 TestClient 進行測試
from fastapi.testclient import TestClient
from main import app
client = TestClient(app)
def test_read_main():
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"msg": "Hello World"}
內容解密:
上述程式碼展示瞭如何使用 TestClient
對 FastAPI 應用進行測試。test_read_main
函式測試根路徑 /
的 GET 請求,驗證回應狀態碼和內容是否符合預期。