INNER CODE UNIT · Python

_normalize_payload

caipe-io/ai-platform-engineering · ai_platform_engineering/audit_service/main.py:173

def _normalize_payload(payload: Any) -> list[dict[str, Any]]:
    if isinstance(payload, list):
        raw_events = payload
    elif isinstance(payload, dict) and isinstance(payload.get("events"), list):
        raw_events = payload["events"]
    elif isinstance(payload, dict):
        raw_events = [payload]
    else:
        raise HTTPException(status_code=400, detail="expected an event object, event array, or {events: [...]}")

    events: list[dict[str, Any]] = []
    for raw in raw_events:
        if not isinstance(raw, dict):
            raise HTTPException(status_code=400, detail="each audit event must be a JSON object")
        events.append(AuditEvent.model_validate(raw).to_record())
    return events

View source record →