feat: implement Phase 2 stabilization
Some checks failed
CI / lint-and-test (push) Has been cancelled
Some checks failed
CI / lint-and-test (push) Has been cancelled
- Cache Graph API tokens with expiry-aware reuse in graph/auth.py - Add tenacity-based retry/backoff wrapper (utils/http.py) and apply to all Graph/source API calls - Add Pydantic request/response models (models/api.py) and FastAPI query constraints - Add unit tests for event_model, auth and integration tests for API endpoints - Configure ruff linter/formatter in pyproject.toml - Add GitHub Actions CI pipeline (.github/workflows/ci.yml) - Add requirements-dev.txt with pytest, mongomock, httpx, ruff - Clean up typing imports and fix ruff linting across codebase
This commit is contained in:
@@ -1,6 +1,5 @@
|
||||
from typing import Dict, List, Optional
|
||||
|
||||
import requests
|
||||
from utils.http import get_with_retry
|
||||
|
||||
|
||||
def _name_from_payload(payload: dict, kind: str) -> str:
|
||||
@@ -26,18 +25,18 @@ def _name_from_payload(payload: dict, kind: str) -> str:
|
||||
return payload.get("displayName") or payload.get("id") or "Unknown"
|
||||
|
||||
|
||||
def _request_json(url: str, token: str) -> Optional[dict]:
|
||||
def _request_json(url: str, token: str) -> dict | None:
|
||||
try:
|
||||
res = requests.get(url, headers={"Authorization": f"Bearer {token}"}, timeout=10)
|
||||
res = get_with_retry(url, headers={"Authorization": f"Bearer {token}"}, timeout=10)
|
||||
if res.status_code == 404:
|
||||
return None
|
||||
res.raise_for_status()
|
||||
return res.json()
|
||||
except requests.RequestException:
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def resolve_directory_object(object_id: str, token: str, cache: Dict[str, dict]) -> Optional[dict]:
|
||||
def resolve_directory_object(object_id: str, token: str, cache: dict[str, dict]) -> dict | None:
|
||||
"""
|
||||
Resolve a directory object (user, servicePrincipal, group, device) to a readable name.
|
||||
Uses a simple multi-endpoint probe with caching to avoid extra Graph traffic.
|
||||
@@ -69,7 +68,7 @@ def resolve_directory_object(object_id: str, token: str, cache: Dict[str, dict])
|
||||
return None
|
||||
|
||||
|
||||
def resolve_service_principal_owners(sp_id: str, token: str, cache: Dict[str, List[str]]) -> List[str]:
|
||||
def resolve_service_principal_owners(sp_id: str, token: str, cache: dict[str, list[str]]) -> list[str]:
|
||||
"""Return a list of owner display names for a service principal."""
|
||||
if not sp_id:
|
||||
return []
|
||||
|
||||
Reference in New Issue
Block a user