apim-apps/apim_apps/client.py

from typing import Optional

import requests

from .models import AppSpec, EnvConfig, OrgSpec


class _AppendOnlySession(requests.Session):
    """HTTP session that refuses DELETE requests — this tool is append-only."""

    def request(self, method: str, url, **kwargs):
        if method.upper() == "DELETE":
            raise RuntimeError(
                f"DELETE is not allowed: this tool never removes existing resources ({url})"
            )
        return super().request(method, url, **kwargs)


class AxwayClient:
    """Axway API Manager REST client. Endpoint: /api/portal/v1.4"""

    def __init__(self, env: EnvConfig):
        self.env = env
        self.base = env.url.rstrip("/") + "/api/portal/v1.4"
        self.session = _AppendOnlySession()
        self.session.auth = (env.username, env.password)
        self.session.verify = env.verify_ssl
        self.session.headers.update({"Accept": "application/json"})

    # ── Helpers ───────────────────────────────────────────────────────────────

    def _get_all(self, path: str, params: dict = None) -> list:
        url = f"{self.base}{path}"
        p = dict(params or {})
        p.setdefault("count", 100)
        p["startIndex"] = 0
        items = []
        while True:
            r = self.session.get(url, params=p, timeout=30)
            r.raise_for_status()
            data = r.json()
            if isinstance(data, list):
                items.extend(data)
                break
            chunk = data.get("result", data.get("items", []))
            if not isinstance(chunk, list):
                return [data]
            items.extend(chunk)
            total = data.get("total", len(chunk))
            p["startIndex"] += len(chunk)
            if p["startIndex"] >= total or not chunk:
                break
        return items

    def _post(self, path: str, payload: dict) -> dict:
        r = self.session.post(
            f"{self.base}{path}",
            json=payload,
            headers={"Content-Type": "application/json"},
            timeout=30,
        )
        r.raise_for_status()
        return r.json()

    # ── Organisations ─────────────────────────────────────────────────────────

    def list_orgs(self) -> dict[str, dict]:
        return {o["name"]: o for o in self._get_all("/organizations")}

    def create_org(self, spec: OrgSpec) -> dict:
        return self._post("/organizations", {
            "name":        spec.name,
            "description": spec.description,
            "email":       spec.email,
            "phone":       spec.phone,
            "enabled":     spec.enabled,
            "development": spec.development,
        })

    # ── Applications ──────────────────────────────────────────────────────────

    def list_apps(self) -> dict[tuple, dict]:
        apps = self._get_all("/applications")
        orgs_by_id = {o["id"]: o["name"] for o in self._get_all("/organizations")}
        return {
            (orgs_by_id.get(a.get("organizationId", ""), "?"), a["name"]): a
            for a in apps
        }

    def create_app(self, spec: AppSpec, org_id: str) -> dict:
        return self._post("/applications", {
            "name":           spec.name,
            "organizationId": org_id,
            "description":    spec.description,
            "email":          spec.email,
            "phone":          spec.phone,
            "enabled":        spec.enabled,
            "state":          "approved",
        })

    def create_app_ext_credential(self, app_id: str, client_id: str, secret: str = "") -> dict:
        payload: dict = {"clientId": client_id, "enabled": True}
        if secret:
            payload["secret"] = secret
        return self._post(f"/applications/{app_id}/extcredentials", payload)

    # ── APIs front-end ────────────────────────────────────────────────────────

    def list_frontend_apis(self) -> list[dict]:
        return self._get_all("/proxies")

    def find_frontend_api(self, api_name: str, api_version: str = "") -> Optional[dict]:
        for api in self.list_frontend_apis():
            if api.get("name", "") == api_name:
                if not api_version or api.get("version", "") == api_version:
                    return api
        return None

    # ── Souscriptions ─────────────────────────────────────────────────────────

    def list_app_api_access(self, app_id: str) -> list[dict]:
        try:
            return self._get_all(f"/applications/{app_id}/apis")
        except requests.HTTPError:
            return []

    def grant_api_access(self, app_id: str, api_id: str) -> dict:
        return self._post(f"/applications/{app_id}/apis/{api_id}", {})