nathan
6337182226
Resolves CRITICAL #1 from code-health-report-2026-04-13.md Changes: - Add tenacity dependency for retry logic - Create lib/resilience.py with: - resilient_http_call decorator (3 retries, exponential backoff 2s→4s→8s) - CircuitBreaker class (opens after 5 consecutive failures) - handle_404_gracefully decorator for safe resource lookups - Apply retry decorators to all HTTP clients: - workday_client.py: get(), raas() - entra_client.py: get(), get_all_pages() - helix_client.py: get(), post() - intune_client.py: get() - lansweeper_client.py: gql() - fedex_client.py: post() - Add graceful degradation to audit tools: - audit_user_drift(): Wrap Workday, AD, Entra calls separately - audit_device_drift(): Wrap Lansweeper, Intune, Helix calls separately - Both now return systems_available and systems_failed fields - Create check_system_health() tool for proactive monitoring - Add comprehensive unit tests for resilience module Benefits: - HTTP clients now automatically retry transient failures (5xx, timeouts) - Circuit breaker prevents hammering failing services (fast-fail after threshold) - Audit tools continue with partial data if some systems unavailable - Health check tool enables proactive system monitoring before bulk audits
48 lines
1.5 KiB
Python
48 lines
1.5 KiB
Python
"""Lansweeper GraphQL API adapter (lib layer)."""
|
|
|
|
from typing import Any
|
|
import httpx
|
|
from config import LansweeperConfig
|
|
from resilience import resilient_http_call
|
|
|
|
|
|
class LansweeperClient:
|
|
"""Low-level Lansweeper Cloud GraphQL client."""
|
|
|
|
def __init__(self):
|
|
self.cfg = LansweeperConfig()
|
|
self._token: str | None = None
|
|
self._http = httpx.AsyncClient(timeout=30)
|
|
|
|
async def get_token(self) -> str:
|
|
if self._token:
|
|
return self._token
|
|
resp = await self._http.post(
|
|
"https://app.lansweeper.com/api/oauth/token",
|
|
json={
|
|
"client_id": self.cfg.application_id,
|
|
"client_secret": self.cfg.application_secret,
|
|
"grant_type": "client_credentials",
|
|
},
|
|
)
|
|
resp.raise_for_status()
|
|
self._token = resp.json()["access_token"]
|
|
return self._token
|
|
|
|
@resilient_http_call(service_name="Lansweeper")
|
|
async def gql(self, query: str, variables: dict | None = None) -> Any:
|
|
token = await self.get_token()
|
|
resp = await self._http.post(
|
|
self.cfg.api_url,
|
|
json={"query": query, "variables": variables or {}},
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if "errors" in payload:
|
|
raise RuntimeError(f"Lansweeper GQL error: {payload['errors']}")
|
|
return payload["data"]
|
|
|
|
async def close(self):
|
|
await self._http.aclose()
|