返回文章列表

FastAPI 深度解析與實務應用

FastAPI 作為新興 Python Web 框架,以高效能、易用性和豐富特性著稱。本文探討 FastAPI 的核心概念、關鍵技術及其在實際專案中的應用,包含路徑操作、引數處理、資料驗證、資料函式庫整合、API 檔案生成、安全性、WebSockets 與測試佈署等導向,協助開發者快速掌握 FastAPI

Web 開發 Python

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 請求,驗證回應狀態碼和內容是否符合預期。