Saltar al contenido principal

TokenProvider

Cachea el token y garantiza una sola autenticación concurrente (single-flight).

  • Sin token, un hilo autentica y los demás esperan y reutilizan el resultado.
  • Ante un 401, cada request informa qué token usó; si otro hilo ya renovó, se reutiliza el nuevo en vez de autenticar otra vez (N hilos con 401 → 1 renovación).

Ejemplo General​

```python
provider = TokenProvider(AuthService(config, httpx.Client()))
token = provider.get_token() # autentica la primera vez
token = provider.get_token() # devuelve el token cacheado
```

Métodos​

def __init__(auth_service: AuthService) -> None​

Crea el proveedor.

Parámetros:

  • auth_service (AuthService) — Servicio que autentica contra Shori.

Ejemplo de uso:

```python
provider = TokenProvider(AuthService(config, httpx.Client()))
```

def get_token() -> str​

Devuelve el token vigente; autentica de forma perezosa si aún no hay.

Retorna: str — Token Bearer.

Lanza:

  • AuthenticationException — Si la autenticación falla.

Ejemplo de uso:

```python
headers = {"Authorization": f"Bearer {provider.get_token()}"}
```

def invalidate_and_refresh(stale_token: str | None) -> str​

Invalida el token y fuerza una nueva autenticación.

Parámetros:

  • stale_token (str | None) — Token que el servidor rechazó. Si el token actual ya es otro, alguien

Retorna: str — Token Bearer vigente.

Ejemplo de uso:

```python
token = provider.get_token()
# ... el servidor responde 401 con ese token ...
token = provider.invalidate_and_refresh(token)
```

def clear() -> None​

Limpia el token cacheado (útil en tests o al cerrar el cliente).

Ejemplo de uso:

```python
provider.clear() # el siguiente get_token() vuelve a autenticar
```