Tomas Kracmar
82bafc06c9
Some checks failed
CI / lint-and-test (push) Has been cancelled
- Fix auth by using idToken fallback when accessToken audience mismatches - Add PyJWT verification with audience-aware token selection in frontend - Source health: track last_attempt_time and error status per source - Frontend: fix modal outside x-data scope, add circular-safe JSON stringify - Frontend: support multi-select service filter with All/None toggles - Frontend: improve filter layout into organized rows - Frontend: fix text overflow and result pill colors (success/succeeded) - Intune: normalize application actors (auditActorType=Application) - Add cache-control middleware for HTML/API responses - Update tests for multi-service filtering and source health
36 lines
1.1 KiB
Python
36 lines
1.1 KiB
Python
|
|
from auth import require_auth
|
|
from fastapi import APIRouter, Depends
|
|
from models.api import SourceHealthResponse
|
|
from watermark import watermarks_collection
|
|
|
|
router = APIRouter(dependencies=[Depends(require_auth)])
|
|
|
|
SOURCES = ["directory", "unified", "intune"]
|
|
|
|
|
|
@router.get("/source-health", response_model=list[SourceHealthResponse])
|
|
def source_health():
|
|
"""Return the last known fetch status for each ingestion source."""
|
|
results = []
|
|
for source in SOURCES:
|
|
doc = watermarks_collection.find_one({"source": source})
|
|
if doc:
|
|
status = doc.get("status")
|
|
if not status:
|
|
status = "healthy" if doc.get("last_fetch_time") else "unknown"
|
|
results.append({
|
|
"source": source,
|
|
"last_fetch_time": doc.get("last_fetch_time"),
|
|
"last_attempt_time": doc.get("last_attempt_time"),
|
|
"status": status,
|
|
})
|
|
else:
|
|
results.append({
|
|
"source": source,
|
|
"last_fetch_time": None,
|
|
"last_attempt_time": None,
|
|
"status": "unknown",
|
|
})
|
|
return results
|