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

內容解密:

  1. 使用 FastAPI() 建立應使用案例項。
  2. 定義 Pydantic 模型 Item,包含 nameprice 欄位。
  3. @app.post("/items/") 裝飾器將 create_item 函式與 POST 請求繫結。
  4. 函式接收 Item 型別的引數,並自動進行資料驗證。
  5. 傳回處理後的 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}

內容解密:

  1. item_id 為路徑引數,自動從 URL 中擷取並轉換為整數型別。
  2. q 為查詢引數,可選,預設值為 None
  3. 函式傳回包含 item_idq 的 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

內容解密:

  1. 定義 User 模型,包含 usernameemail 欄位。
  2. 使用 @validator 裝飾器對 email 欄位進行自定義驗證。
  3. 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

內容解密:

  1. 建立非同步資料函式庫引擎和會話管理器。
  2. 定義 get_db 依賴函式,提供資料函式庫會話。
  3. 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",
)

內容解密:

  1. 在建立 FastAPI 應用時提供元資料資訊。
  2. 自動生成的 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}

內容解密:

  1. 建立 OAuth2 認證方案,指定 token 取得 URL。
  2. 在受保護的路由中使用 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 請求,驗證回應狀態碼和內容是否符合預期。