pulp-console/app/pulp_client.py

"""Pulp REST API 호출 래퍼 (스펙 §7).

브라우저가 아니라 이 BFF 가 Basic Auth 를 쥐고 Pulp 를 호출한다(스펙 §2).
객체는 이름이 아니라 ``pulp_href``(경로 문자열) 기준으로 참조한다(스펙 §4).

폴링은 화면(HTMX)이 담당하므로 여기서는 task 단발 조회만 둔다. sync/deploy 등
도메인 메서드는 이후 단계에서 추가한다.
"""

from __future__ import annotations

import time
from typing import Any

import httpx

from .config import Settings, get_settings, tls_verify

API_PREFIX = "/pulp/api/v3"


class PulpClient:
    def __init__(self, settings: Settings | None = None) -> None:
        self.settings = settings or get_settings()
        self._client = httpx.Client(
            base_url=self.settings.pulp_base_url.rstrip("/"),
            auth=(self.settings.pulp_username, self.settings.pulp_password),
            verify=tls_verify(self.settings),
            timeout=30.0,
        )

    def close(self) -> None:
        self._client.close()

    def __enter__(self) -> PulpClient:
        return self

    def __exit__(self, *exc: object) -> None:
        self.close()

    # --- 저수준 호출 (Basic Auth 포함) ---------------------------------
    def get(self, path: str, **kwargs: Any) -> dict[str, Any]:
        resp = self._client.get(path, **kwargs)
        resp.raise_for_status()
        return resp.json()

    def post(self, path: str, json: Any | None = None, **kwargs: Any) -> dict[str, Any]:
        resp = self._client.post(path, json=json, **kwargs)
        resp.raise_for_status()
        return resp.json()

    def patch(
        self, href: str, json: Any | None = None, **kwargs: Any
    ) -> dict[str, Any]:
        resp = self._client.patch(href, json=json, **kwargs)
        resp.raise_for_status()
        return resp.json()

    # --- 도메인 헬퍼 ----------------------------------------------------
    def status(self) -> dict[str, Any]:
        """헬스체크: GET /pulp/api/v3/status/"""
        return self.get(f"{API_PREFIX}/status/")

    def list_repos(self) -> list[dict[str, Any]]:
        """RPM 저장소 목록: GET /pulp/api/v3/repositories/rpm/rpm/ → results."""
        data = self.get(f"{API_PREFIX}/repositories/rpm/rpm/")
        return data.get("results", [])

    def get_version(self, version_href: str) -> dict[str, Any]:
        """RepositoryVersion 단건 조회 (href 기준). content_summary 등 포함."""
        return self.get(version_href)

    def list_versions(self, uuid: str) -> list[dict[str, Any]]:
        """저장소의 RepositoryVersion 목록: GET .../{uuid}/versions/ → results."""
        data = self.get(f"{API_PREFIX}/repositories/rpm/rpm/{uuid}/versions/")
        return data.get("results", [])

    def list_distributions(self) -> list[dict[str, Any]]:
        """Distribution 목록: GET /pulp/api/v3/distributions/rpm/rpm/ → results.

        운영자가 yum/dnf baseurl 로 바라보는 공개 URL들. publication 으로 배포 버전 추적.
        """
        data = self.get(f"{API_PREFIX}/distributions/rpm/rpm/")
        return data.get("results", [])

    def get_publication(self, publication_href: str) -> dict[str, Any]:
        """Publication 단건 조회 (repository_version 역추적용)."""
        return self.get(publication_href)

    def get_repo(self, uuid: str) -> dict[str, Any]:
        """저장소 단건 조회 (remote href 등 확인용)."""
        return self.get(f"{API_PREFIX}/repositories/rpm/rpm/{uuid}/")

    def sync_repo(self, uuid: str, remote_href: str) -> str:
        """동기화 시작 → 비동기 task. 반환 task_href 를 폴링으로 추적한다(스펙 §4).

        POST .../repositories/rpm/rpm/{uuid}/sync/  body {"remote": <remote_href>}
        응답 {"task": "/pulp/api/v3/tasks/<uuid>/"} 에서 task href 추출.
        """
        data = self.post(
            f"{API_PREFIX}/repositories/rpm/rpm/{uuid}/sync/",
            json={"remote": remote_href},
        )
        return data["task"]

    def get_task(self, task_href: str) -> dict[str, Any]:
        """task 단건 조회 (state, progress_reports, created_resources, error)."""
        return self.get(task_href)

    def create_publication(self, version_href: str) -> str:
        """특정 버전을 배포 가능한 형태로: POST /publications/rpm/rpm/ → task_href.

        완료 후 task.created_resources 에서 publication href 를 얻는다.
        """
        data = self.post(
            f"{API_PREFIX}/publications/rpm/rpm/",
            json={"repository_version": version_href},
        )
        return data["task"]

    def update_distribution(self, dist_href: str, publication_href: str) -> str:
        """배포 확정: PATCH {dist_href} publication 교체 → task_href.

        운영망이 보는 URL 이 이 publication 을 가리키게 된다(실제 배포 동작, 스펙 §4).
        """
        data = self.patch(dist_href, json={"publication": publication_href})
        return data["task"]

    def wait_for_task(
        self, task_href: str, attempts: int = 120, delay: float = 1.0
    ) -> dict[str, Any]:
        """task 가 종료 상태(completed/failed/canceled)가 될 때까지 대기 후 반환.

        배포는 (publication 생성 → distribution 교체) 2단계 task 라 결과 확정이 필요해
        서버측에서 짧게 대기한다. 화면 폴링이 어려운 복합 동작에 한정해서 쓴다.
        """
        terminal = {"completed", "failed", "canceled"}
        task = self.get_task(task_href)
        for _ in range(attempts):
            if task.get("state") in terminal:
                return task
            time.sleep(delay)
            task = self.get_task(task_href)
        return task