soc/soc-integrator/app/adapters/iris.py

from typing import Any

import httpx


class IrisAdapter:
    def __init__(self, base_url: str, api_key: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key

    def _headers(self) -> dict[str, str]:
        return {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}

    async def create_case(self, payload: dict[str, Any]) -> dict[str, Any]:
        headers = self._headers()
        async with httpx.AsyncClient(verify=False, timeout=20.0) as client:
            v2_url = f"{self.base_url}/api/v2/cases"
            response = await client.post(v2_url, json=payload, headers=headers)

            if response.status_code == 404:
                legacy_url = f"{self.base_url}/manage/cases/add"
                response = await client.post(legacy_url, json=payload, headers=headers)
                active_url = legacy_url
            else:
                active_url = v2_url

            try:
                response.raise_for_status()
            except httpx.HTTPStatusError as exc:
                detail = response.text.strip()
                raise RuntimeError(
                    f"IRIS returned {response.status_code} for {active_url}. Response: {detail}"
                ) from exc
            return response.json() if response.content else {"status_code": response.status_code}

    async def update_case(self, case_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        headers = self._headers()
        async with httpx.AsyncClient(verify=False, timeout=20.0) as client:
            v2_url = f"{self.base_url}/api/v2/cases/{case_id}"
            response = await client.put(v2_url, json=payload, headers=headers)

            if response.status_code == 404:
                legacy_url = f"{self.base_url}/manage/cases/update/{case_id}"
                response = await client.post(legacy_url, json=payload, headers=headers)
                active_url = legacy_url
            else:
                active_url = v2_url

            try:
                response.raise_for_status()
            except httpx.HTTPStatusError as exc:
                detail = response.text.strip()
                raise RuntimeError(
                    f"IRIS returned {response.status_code} for {active_url}. Response: {detail}"
                ) from exc
            return response.json() if response.content else {"status_code": response.status_code}

    async def whoami(self) -> dict[str, Any]:
        headers = self._headers()
        url = f"{self.base_url}/user/whoami"
        async with httpx.AsyncClient(verify=False, timeout=20.0) as client:
            response = await client.get(url, headers=headers)
            response.raise_for_status()
            return response.json() if response.content else {"status_code": response.status_code}

    async def list_cases(self, limit: int = 50, offset: int = 0) -> dict[str, Any]:
        headers = self._headers()
        url = f"{self.base_url}/api/v2/cases"
        params = {"limit": max(1, limit), "offset": max(0, offset)}
        async with httpx.AsyncClient(verify=False, timeout=20.0) as client:
            response = await client.get(url, params=params, headers=headers)
            try:
                response.raise_for_status()
            except httpx.HTTPStatusError as exc:
                detail = response.text.strip()
                raise RuntimeError(
                    f"IRIS returned {response.status_code} for {url}. Response: {detail}"
                ) from exc
            return response.json() if response.content else {"status_code": response.status_code}