xiaoxia-saas/packages/application/auth/jwt_handler.py

"""
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", role="admin")
    payload = jwt_handler.verify_access_token(token)
"""

from typing import Any, Dict, Optional

from packages.application.auth.jwt_service import JWTConfig, JWTService


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,
        role: str = "",
        additional_claims: Optional[Dict[str, Any]] = None,
    ) -> str:
        """
        创建 access_token

        Args:
            user_id: 用户 ID
            role: 用户角色
            additional_claims: 额外的声明信息

        Returns:
            JWT Token 字符串
        """
        return self._service.create_access_token(
            user_id=user_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