动态身份验证

1
from .client import PlantStoreClient as FernClient
2
import jwt
3
import time
4
from typing import Optional, Dict, Any
5
6
class PlantStoreClient(FernClient):
7
    def __init__(self, *, private_key: str, environment: str):
8
        super().__init__(environment=environment)
9
        self._private_key = private_key
10
    
11
    def _generate_jwt(self) -> str:
12
        """生成有效期为 15 秒的短期 JWT 令牌"""
13
        now = int(time.time())
14
        payload = {
15
            "iat": now,
16
            "exp": now + 15,
17
        }
18
        return jwt.encode(payload, self._private_key, algorithm="RS256")
19
    
20
    def _with_jwt(self, request_options: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
21
        """将 JWT 注入请求选项"""
22
        token = self._generate_jwt()
23
        options = request_options or {}
24
        headers = options.get("headers", {})
25
        headers["Authorization"] = f"Bearer {token}"
26
        options["headers"] = headers
27
        return options
28
    
29
    def get_plant(self, plant_id: str, *, request_options: Optional[Dict[str, Any]] = None):
30
        return super().plants.get(plant_id, request_options=self._with_jwt(request_options))
31
    
32
    def create_plant(self, request: Any, *, request_options: Optional[Dict[str, Any]] = None):
33
        return super().plants.create(request, request_options=self._with_jwt(request_options))
34
    
35
    # 根据需要重写其他方法...

1
from .wrapper.client import PlantStoreClient
2
3
__all__ = ["PlantStoreClient"]

1
+ src/wrapper
2
+ src/__init__.py