Раис Юсупалиев
98 lines
3.4 KiB
Python
98 lines
3.4 KiB
Python
"""CDEK OAuth2 authentication adapter."""
|
|
|
|
import asyncio
|
|
import time
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
|
|
class CDEKAuthError(RuntimeError):
|
|
"""Raised when CDEK OAuth2 flow fails."""
|
|
|
|
|
|
class CDEKAuthClient:
|
|
def __init__(
|
|
self,
|
|
http_client: httpx.AsyncClient,
|
|
*,
|
|
base_url: str,
|
|
client_id: str,
|
|
client_secret: str,
|
|
timeout_seconds: float = 10.0,
|
|
expiry_skew_seconds: float = 30.0,
|
|
clock: Callable[[], float] | None = None,
|
|
) -> None:
|
|
self._http_client = http_client
|
|
self._token_url = f"{base_url.rstrip('/')}/oauth/token"
|
|
self._client_id = client_id
|
|
self._client_secret = client_secret
|
|
self._timeout_seconds = timeout_seconds
|
|
self._expiry_skew_seconds = expiry_skew_seconds
|
|
self._clock = clock or time.monotonic
|
|
self._token: str | None = None
|
|
self._token_expires_at: float = 0.0
|
|
self._refresh_lock = asyncio.Lock()
|
|
|
|
async def get_access_token(self) -> str:
|
|
if self._has_valid_token():
|
|
return self._cached_token()
|
|
|
|
async with self._refresh_lock:
|
|
if self._has_valid_token():
|
|
return self._cached_token()
|
|
return await self._fetch_access_token()
|
|
|
|
def _has_valid_token(self) -> bool:
|
|
return self._token is not None and self._clock() < self._token_expires_at
|
|
|
|
def _cached_token(self) -> str:
|
|
if self._token is None:
|
|
raise CDEKAuthError("CDEK OAuth token is unexpectedly missing.")
|
|
return self._token
|
|
|
|
async def _fetch_access_token(self) -> str:
|
|
try:
|
|
response = await self._http_client.post(
|
|
self._token_url,
|
|
data={
|
|
"grant_type": "client_credentials",
|
|
"client_id": self._client_id,
|
|
"client_secret": self._client_secret,
|
|
},
|
|
timeout=self._timeout_seconds,
|
|
)
|
|
response.raise_for_status()
|
|
payload = response.json()
|
|
token = self._extract_access_token(payload)
|
|
expires_in = self._extract_expires_in(payload)
|
|
except (httpx.HTTPError, TypeError, ValueError) as exc:
|
|
raise CDEKAuthError("Failed to obtain CDEK OAuth token.") from exc
|
|
|
|
self._token = token
|
|
ttl_seconds = max(expires_in - self._expiry_skew_seconds, 0.0)
|
|
self._token_expires_at = self._clock() + ttl_seconds
|
|
return self._token
|
|
|
|
@staticmethod
|
|
def _extract_access_token(payload: Any) -> str:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("CDEK OAuth payload must be a JSON object.")
|
|
token = payload.get("access_token")
|
|
if not isinstance(token, str) or not token.strip():
|
|
raise ValueError("CDEK OAuth payload has no valid access_token.")
|
|
return token
|
|
|
|
@staticmethod
|
|
def _extract_expires_in(payload: Any) -> float:
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("CDEK OAuth payload must be a JSON object.")
|
|
expires_in = payload.get("expires_in")
|
|
if expires_in is None:
|
|
raise ValueError("CDEK OAuth payload has no expires_in.")
|
|
parsed_expires_in = float(expires_in)
|
|
if parsed_expires_in < 0:
|
|
raise ValueError("CDEK OAuth expires_in cannot be negative.")
|
|
return parsed_expires_in
|