je_load_density.utils.auth

class je_load_density.utils.auth.OAuth2Client(token_url: str, client_id: str, client_secret: str, scope: Optional[str] = None, timeout: float = 5.0, safety_window: float = 30.0, poster: Callable[[str, bytes, Dict[str, str], float], Dict[str, Any]] = <function _default_poster at 0x711191e8bf60>, _cache: Dict[Tuple[str, str], je_load_density.utils.auth.oauth2._CacheEntry] = <factory>, _lock: <built-in function allocate_lock> = <factory>)

Bases: object

clear() → None

client_id: str

client_secret: str

get_client_credentials() → Dict[str, Any]

get_password(username: str, password: str) → Dict[str, Any]

poster(body: bytes, headers: Dict[str, str], timeout: float) → Dict[str, Any]

refresh(refresh: str) → Dict[str, Any]

safety_window: float = 30.0

scope: str | None = None

timeout: float = 5.0

token_url: str

je_load_density.utils.auth.decode_jwt(token: str) → Tuple[Dict[str, Any], Dict[str, Any], bytes]: Decode the three JWT segments without verifying the signature. Returns (header_dict, payload_dict, signature_bytes).

je_load_density.utils.auth.fetch_client_credentials_token(token_url: str, client_id: str, client_secret: str, scope: str | None = None, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.utils.auth.fetch_password_token(token_url: str, client_id: str, client_secret: str, username: str, password: str, scope: str | None = None, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.utils.auth.refresh_token(token_url: str, client_id: str, client_secret: str, refresh: str, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.utils.auth.sign_aws_request(method: str, url: str, region: str, service: str, access_key: str, secret_key: str, body: bytes | str = b'', headers: Mapping[str, str] | None = None, session_token: str | None = None, now: datetime | None = None) → Dict[str, str]: Return a new headers dict carrying the SigV4 Authorization header, x-amz-date, optional x-amz-security-token, and x-amz-content-sha256.

je_load_density.utils.auth.sign_jwt(payload: Dict[str, Any], secret: str | bytes, algorithm: str = 'HS256', headers: Dict[str, Any] | None = None, expires_in_seconds: int | None = None) → str: Build and sign a JSON Web Token. Mutates a copy of payload — if expires_in_seconds is set, exp is added.

Modules

`aws_sigv4`	AWS Signature Version 4 signer.
`jwt_signer`	JWT signer (HS256, HS384, HS512, RS256, RS384, RS512).
`oauth2`	OAuth2 token helpers with a tiny in-memory cache.