From e70d3280872ffa06bf3084cc45f29fbe937c6bee Mon Sep 17 00:00:00 2001
From: xiaoxia <dev@xiaoxiajianji.com>
Date: Fri, 26 Jun 2026 18:35:45 +0800
Subject: [PATCH] =?UTF-8?q?P2-2:=20=E5=88=9B=E5=BB=BA=20packages/applicati?=
 =?UTF-8?q?on/auth/jwt=5Fhandler.py=20=E5=A7=94=E6=89=98=E5=B1=82=EF=BC=8C?=
 =?UTF-8?q?=E9=9A=94=E7=A6=BB=20Domain=20=E5=B1=82=E5=AF=B9=20jwt=20?=
 =?UTF-8?q?=E5=BA=93=E7=9A=84=E4=BE=9D=E8=B5=96?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 packages/application/auth/jwt_handler.py | 143 +++++++++++++++++++++++
 1 file changed, 143 insertions(+)
 create mode 100644 packages/application/auth/jwt_handler.py

diff --git a/packages/application/auth/jwt_handler.py b/packages/application/auth/jwt_handler.py
new file mode 100644
index 000000000..7c0efef84
--- /dev/null
+++ b/packages/application/auth/jwt_handler.py
@@ -0,0 +1,143 @@
+"""
+JWT 处理器委托层
+
+此模块作为 Domain 层 (packages.domain.auth.jwt_service) 和 Application 层之间的委托层，
+隔离 Domain 层对 jwt 库的直接依赖。
+
+使用方式：
+    from packages.application.auth.jwt_handler import JWTHandler, get_jwt_handler
+    
+    jwt_handler = JWTHandler(secret_key="your-secret-key")
+    token = jwt_handler.create_access_token(user_id="user123", workspace_id="ws456", role="admin")
+    payload = jwt_handler.verify_access_token(token)
+"""
+
+from datetime import datetime, timedelta
+from typing import Any, Dict, Optional
+
+from packages.domain.auth.jwt_service import JWTConfig, JWTService, TokenType
+
+
+class JWTHandler:
+    """
+    JWT 处理器委托类
+    
+    委托给 packages.domain.auth.jwt_service.JWTService 进行实际的 JWT 操作，
+    此层仅负责配置和封装，不直接依赖 jwt 库。
+    """
+
+    def __init__(self, secret_key: str, algorithm: str = "HS256", access_token_expire_minutes: int = 30):
+        """
+        初始化 JWT 处理器
+        
+        Args:
+            secret_key: JWT 签名密钥（必须从环境变量或配置注入）
+            algorithm: 加密算法，默认 HS256
+            access_token_expire_minutes: Access Token 过期时间（分钟）
+        """
+        config = JWTConfig(
+            secret_key=secret_key,
+            algorithm=algorithm,
+            access_token_expire_minutes=access_token_expire_minutes,
+        )
+        self._service = JWTService(config)
+
+    def create_access_token(
+        self,
+        user_id: str,
+        workspace_id: str,
+        role: str,
+        additional_claims: Optional[Dict[str, Any]] = None,
+    ) -> str:
+        """
+        创建 access_token
+        
+        Args:
+            user_id: 用户 ID
+            workspace_id: 工作空间 ID
+            role: 用户角色
+            additional_claims: 额外的声明信息
+            
+        Returns:
+            JWT Token 字符串
+        """
+        return self._service.create_access_token(
+            user_id=user_id,
+            workspace_id=workspace_id,
+            role=role,
+            additional_claims=additional_claims,
+        )
+
+    def verify_access_token(self, token: str) -> Dict[str, Any]:
+        """
+        验证 access_token
+        
+        Args:
+            token: JWT Token 字符串
+            
+        Returns:
+            Token payload
+            
+        Raises:
+            ExpiredSignatureError: Token 已过期
+            ValueError: Token 类型不是 access
+        """
+        return self._service.verify_access_token(token)
+
+    def verify_token(self, token: str) -> Dict[str, Any]:
+        """
+        验证任意 Token
+        
+        Args:
+            token: JWT Token 字符串
+            
+        Returns:
+            Token payload
+        """
+        return self._service.verify_token(token)
+
+
+# 默认处理器实例（需要通过 configure_jwt_handler 配置）
+_default_handler: Optional[JWTHandler] = None
+
+
+def configure_jwt_handler(
+    secret_key: str,
+    algorithm: str = "HS256",
+    access_token_expire_minutes: int = 30,
+) -> JWTHandler:
+    """
+    配置全局 JWT 处理器
+    
+    Args:
+        secret_key: JWT 签名密钥
+        algorithm: 加密算法
+        access_token_expire_minutes: Access Token 过期时间（分钟）
+        
+    Returns:
+        配置好的 JWTHandler 实例
+    """
+    global _default_handler
+    _default_handler = JWTHandler(
+        secret_key=secret_key,
+        algorithm=algorithm,
+        access_token_expire_minutes=access_token_expire_minutes,
+    )
+    return _default_handler
+
+
+def get_jwt_handler() -> JWTHandler:
+    """
+    获取全局 JWT 处理器
+    
+    Returns:
+        JWTHandler 实例
+        
+    Raises:
+        RuntimeError: 如果尚未配置 JWT 处理器
+    """
+    if _default_handler is None:
+        raise RuntimeError(
+            "JWT handler not configured. Call configure_jwt_handler() first."
+        )
+    return _default_handler