timmy / FastAPI Token 認證服務腳本

# /// script # requires-python = ">=3.12" # dependencies = [ # "fastapi", # "uvicorn", # "pydantic", # ] # /// import logging import uvicorn from typing import Annotated, Dict, Optional from fastapi import FastAPI, Depends, HTTPException, status from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from pydantic import BaseModel # ----------------------------------------------------------------------------- # 1. 基礎設定與 Mock 資料 (Mock Data) # ----------------------------------------------------------------------------- # 設定 Logger，分級顯示資訊 logging.basicConfig( level=logging.INFO, format="%(asctime)s - [%(levelname)s] - %(message)s" ) logger = logging.getLogger("AuthSystem") # 假資料：使用者資料庫 (Table-driven lookup) # 使用 Dict 作為查找表，取代資料庫查詢 MOCK_USER_DB: Dict[str, Dict[str, str]] = { "admin-secret-token": { "username": "admin_user", "email": "admin@example.com", "role": "admin" }, "guest-access-token": { "username": "guest_user", "email": "guest@example.com", "role": "guest" } } # ----------------------------------------------------------------------------- # 2. 資料模型 (Pydantic Models) # ----------------------------------------------------------------------------- class UserProfile(BaseModel): """回傳給前端的使用者資料模型""" username: str email: str role: str class SystemStatus(BaseModel): """系統狀態模型""" status: str active_users: int # ----------------------------------------------------------------------------- # 3. 核心邏輯類別 (Service Class) # ----------------------------------------------------------------------------- class AuthService: """ 處理認證與授權的服務類別。 將邏輯封裝在此，避免 API 路由函式過於肥大。 """ def __init__(self): # FastAPI 的 HTTPBearer 會自動檢查 Authorization header 是否存在 # auto_error=True 會在 header 缺失時自動回傳 403/401 self.security_scheme = HTTPBearer(auto_error=True) def _get_user_from_db(self, token: str) -> Optional[Dict[str, str]]: """ 模擬從資料庫撈取使用者。 Args: token: 用戶提供的 Bearer Token Returns: Dict: 使用者資料或是 None """ return MOCK_USER_DB.get(token) def _validate_credentials(self, token: str) -> UserProfile: """ 驗證 Token 的有效性並轉換為 Pydantic 模型。 Raises: HTTPException: 當 Token 無效時拋出 401 """ user_data = self._get_user_from_db(token) # 使用 Guard Clauses (衛語句) 避免深層巢狀分支 if not user_data: logger.warning(f"收到無效的 Token 嘗試存取: {token}") raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="無效的憑證 (Invalid Token)", headers={"WWW-Authenticate": "Bearer"}, ) logger.info(f"使用者驗證成功: {user_data['username']}") return UserProfile(**user_data) def __call__(self, creds: Annotated[HTTPAuthorizationCredentials, Depends(HTTPBearer())]) -> UserProfile: """ 使類別實例可被呼叫 (Callable)，作為 FastAPI 的依賴項 (Dependency)。 Args: creds: FastAPI 自動解析的 Bearer 憑證物件 Returns: UserProfile: 驗證通過的使用者物件 """ token = creds.credentials logger.debug(f"開始驗證 Token: {token[:5]}***") # 遮罩敏感資訊 return self._validate_credentials(token) # ----------------------------------------------------------------------------- # 4. FastAPI 應用程式與路由 # ----------------------------------------------------------------------------- # 初始化服務實例 auth_service = AuthService() app = FastAPI( title="FastAPI HTTPBearer Example", description="示範使用 Class-based 依賴注入進行 Token 驗證", version="1.0.0" ) @app.get("/me", response_model=UserProfile) def read_current_user(current_user: Annotated[UserProfile, Depends(auth_service)]): """ 取得當前登入使用者的資訊。 - 需要 Bearer Token 授權 - 根據 Token 回傳對應的 Mock Data """ logger.info(f"API /me 被呼叫，使用者: {current_user.username}") return current_user @app.get("/status", response_model=SystemStatus) def read_system_status(current_user: Annotated[UserProfile, Depends(auth_service)]): """ 取得系統狀態 (僅示範用)。 """ # 這裡可以用 switch/match 處理不同角色的權限 (表驅動概念) # Python 3.10+ match 語法 match current_user.role: case "admin": msg = "系統運作正常 (Admin View)" case _: msg = "系統運作正常" return SystemStatus(status=msg, active_users=len(MOCK_USER_DB)) # ----------------------------------------------------------------------------- # 5. 程式進入點 # ----------------------------------------------------------------------------- if __name__ == "__main__": # 使用 uvicorn 啟動服務 # host="0.0.0.0" 允許外部連線 logger.info("正在啟動 FastAPI 服務...") uvicorn.run(app, host="127.0.0.1", port=8000)