Source: factory.core/ObjKeycloakResilient.py
Circuit breaker states.
Resilient Keycloak client with circuit breaker and token caching.
Provides fault-tolerant authentication that continues working when
Keycloak is unavailable using cached tokens and sync queue.
| Method | Signature | Description |
|---|---|---|
| is_available | is_available() -> bool |
Check if Keycloak is available (respects circuit breaker). |
| authenticate | authenticate(username: str, password: str, package: Optional[str] = None) -> Optional[Dict[str, Any]] |
Authenticate user with Keycloak (best effort). |
| get_cached_token | get_cached_token(username: str, package: str) -> Optional[Dict[str, Any]] |
Get cached Keycloak token if within grace period. |
| refresh_cached_token | refresh_cached_token(username: str, package: str) -> Optional[Dict[str, Any]] |
Attempt to refresh cached token using refresh_token. |
| auto_refresh_expiring_tokens | auto_refresh_expiring_tokens(hours_before_expiry: int = 2) -> int |
Automatically refresh tokens expiring soon. |
| queue_sync | queue_sync(username: str, package: str, sync_type: str, sync_data: Dict[str, Any], priority: int = 5) -> Optional[str] |
Queue a sync operation for when Keycloak recovers. |
| process_sync_queue | process_sync_queue(batch_size: int = 100) -> int |
Process pending sync queue items. |
| provision_users_batch | provision_users_batch(users: List[Dict[str, Any]], package: str) -> Dict[str, Any] |
Provision multiple users to Keycloak in batch. |
| sync_users_from_database | sync_users_from_database(package: str, where_clause: str = 'Active = 1') -> Dict[str, Any] |
Sync all users from sys_user database to Keycloak. |
| get_status | get_status() -> Dict[str, Any] |
Get current Keycloak resilience status. |
| cleanup_expired_tokens | cleanup_expired_tokens(hours: int = 48) -> int |
Remove expired cached tokens. |
| cleanup_completed_syncs | cleanup_completed_syncs(days: int = 7) -> int |
Remove completed sync queue items. |
Show Keycloak resilience status.