Hyemin Lee
72c75252d3
- pulp_client: list_versions / list_distributions / get_publication
- GET /repos/{uuid}/versions: 버전 목록 페이지(최신순)
- 현재 운영 배포 버전 판별: Distribution→publication→repository_version 역추적
- views.build_version_view: vN, 생성일시, 패키지 수, 증감(+/-), 검증 배지, 배포 여부
- views.verification_detail: gpgcheck/repo-gpgcheck/checksum 상세 (화면4)
- 대시보드 저장소명 → 버전 페이지 링크, [이 버전 배포] 버튼은 단계4용 placeholder
- 단계1 TODO(배포 버전 역추적) 해소
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
108 lines
4.1 KiB
Python
108 lines
4.1 KiB
Python
"""Pulp REST API 호출 래퍼 (스펙 §7).
|
|
|
|
브라우저가 아니라 이 BFF 가 Basic Auth 를 쥐고 Pulp 를 호출한다(스펙 §2).
|
|
객체는 이름이 아니라 ``pulp_href``(경로 문자열) 기준으로 참조한다(스펙 §4).
|
|
|
|
폴링은 화면(HTMX)이 담당하므로 여기서는 task 단발 조회만 둔다. sync/deploy 등
|
|
도메인 메서드는 이후 단계에서 추가한다.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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)
|