kjh2064
c3e5eabe90
- tools/gitea/__init__.py: 패키지 진입점, 토큰 우선순위 문서화 - tools/gitea/client.py: GiteaClient (SOLID SRP) - runs/jobs/secrets/vars/PR/releases API - tools/gitea/harness.py: CLI 하네스 - health|runs|run|secrets|vars|workflows|dispatch - AGENTS.md: tools/gitea/ 디렉토리 라우팅 항목 추가 - 검증: health PASS, secrets 6건 확인, ci_lint PASS
316 lines
9.9 KiB
Python
316 lines
9.9 KiB
Python
"""
|
|
tools/gitea/client.py
|
|
Gitea API 공통 클라이언트 - SOLID SRP 준수, 하나의 책임: Gitea REST API 통신
|
|
|
|
formula_id: GITEA_API_CLIENT_V1
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.request
|
|
from typing import Any
|
|
|
|
DEFAULT_BASE_URL = "https://gitea.taxbaik.com"
|
|
DEFAULT_OWNER = "kjh2064"
|
|
DEFAULT_REPO = "QuantEngineByItz"
|
|
|
|
# 토큰 환경변수 우선순위 (높은 것부터)
|
|
_TOKEN_ENV_PRIORITY = (
|
|
"GITEA_TOKEN_BAIK", # 사용자 지정 별칭
|
|
"GITEA_TOKEN_TAXBAIK", # 기존 표준 (현재 환경 보유)
|
|
"GITEA_TOKEN", # 일반 fallback
|
|
"GITEA_TOKEN_HOME", # 레거시
|
|
)
|
|
|
|
|
|
class GiteaApiError(Exception):
|
|
"""Gitea API 호출 오류"""
|
|
def __init__(self, status: int, reason: str, body: Any) -> None:
|
|
self.status = status
|
|
self.reason = reason
|
|
self.body = body
|
|
super().__init__(f"HTTP {status} {reason}: {body}")
|
|
|
|
|
|
def _resolve_token(explicit: str = "") -> str:
|
|
"""
|
|
토큰 해석 우선순위:
|
|
1. explicit 인자 (CLI --token 등)
|
|
2. GITEA_TOKEN_BAIK
|
|
3. GITEA_TOKEN_TAXBAIK
|
|
4. GITEA_TOKEN
|
|
5. GITEA_TOKEN_HOME
|
|
"""
|
|
if explicit and explicit.strip():
|
|
return explicit.strip()
|
|
for key in _TOKEN_ENV_PRIORITY:
|
|
val = os.environ.get(key, "").strip()
|
|
if val:
|
|
return val
|
|
return ""
|
|
|
|
|
|
class GiteaClient:
|
|
"""
|
|
Gitea REST API v1 클라이언트
|
|
|
|
사용 예시:
|
|
client = GiteaClient() # 환경변수 자동 탐지
|
|
client = GiteaClient(token="my_token") # 명시적 토큰
|
|
client = GiteaClient(base_url="https://my.gitea") # 다른 인스턴스
|
|
|
|
모든 메서드는 dict/list를 반환하며, HTTP 오류 시 GiteaApiError를 발생시킵니다.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
base_url: str = DEFAULT_BASE_URL,
|
|
owner: str = DEFAULT_OWNER,
|
|
repo: str = DEFAULT_REPO,
|
|
token: str = "",
|
|
timeout: int = 30,
|
|
) -> None:
|
|
self.base_url = base_url.rstrip("/")
|
|
self.owner = owner
|
|
self.repo = repo
|
|
self.token = _resolve_token(token)
|
|
self.timeout = timeout
|
|
self._api = f"{self.base_url}/api/v1"
|
|
self._repo_url = f"{self._api}/repos/{self.owner}/{self.repo}"
|
|
|
|
# ------------------------------------------------------------------
|
|
# 내부 HTTP 레이어
|
|
# ------------------------------------------------------------------
|
|
|
|
def _request(
|
|
self,
|
|
url: str,
|
|
method: str = "GET",
|
|
body: dict | None = None,
|
|
) -> Any:
|
|
"""HTTP 요청 → JSON 반환. 오류 시 GiteaApiError 발생."""
|
|
headers: dict[str, str] = {
|
|
"Accept": "application/json",
|
|
"User-Agent": "QuantEngine-Gitea-Client/1.0",
|
|
}
|
|
if self.token:
|
|
headers["Authorization"] = f"token {self.token}"
|
|
|
|
data: bytes | None = None
|
|
if body is not None:
|
|
headers["Content-Type"] = "application/json"
|
|
data = json.dumps(body).encode("utf-8")
|
|
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
|
|
raw = resp.read().decode("utf-8", errors="replace")
|
|
return json.loads(raw) if raw else None
|
|
except urllib.error.HTTPError as exc:
|
|
raw = exc.read().decode("utf-8", errors="replace")
|
|
try:
|
|
payload = json.loads(raw)
|
|
except Exception:
|
|
payload = raw
|
|
raise GiteaApiError(exc.code, exc.reason or "", payload)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Repository
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_repo(self) -> dict:
|
|
"""저장소 정보 조회"""
|
|
return self._request(self._repo_url)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Actions Runs
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_runs(self, limit: int = 20, page: int = 1) -> list[dict]:
|
|
"""
|
|
최근 Actions 실행 목록 조회
|
|
|
|
Args:
|
|
limit: 반환할 최대 개수 (기본 20)
|
|
page: 페이지 번호 (기본 1)
|
|
|
|
Returns:
|
|
workflow_run 목록 (list[dict])
|
|
"""
|
|
url = f"{self._repo_url}/actions/runs?limit={limit}&page={page}"
|
|
payload = self._request(url)
|
|
if isinstance(payload, dict):
|
|
return payload.get("workflow_runs") or []
|
|
return []
|
|
|
|
def get_run(self, run_id: int) -> dict:
|
|
"""
|
|
특정 Actions 실행 상세 조회
|
|
|
|
Args:
|
|
run_id: Gitea Actions run ID
|
|
|
|
Returns:
|
|
workflow_run dict
|
|
"""
|
|
url = f"{self._repo_url}/actions/runs/{run_id}"
|
|
return self._request(url)
|
|
|
|
def list_run_jobs(self, run_id: int) -> list[dict]:
|
|
"""
|
|
특정 실행의 job 목록 조회
|
|
|
|
Args:
|
|
run_id: Gitea Actions run ID
|
|
|
|
Returns:
|
|
workflow_job 목록 (list[dict])
|
|
"""
|
|
url = f"{self._repo_url}/actions/runs/{run_id}/jobs"
|
|
payload = self._request(url)
|
|
if isinstance(payload, dict):
|
|
return payload.get("workflow_jobs") or []
|
|
return []
|
|
|
|
def list_workflows(self) -> list[dict]:
|
|
"""
|
|
저장소 workflow 목록 조회
|
|
|
|
Returns:
|
|
workflow 목록 (list[dict])
|
|
"""
|
|
url = f"{self._repo_url}/actions/workflows"
|
|
payload = self._request(url)
|
|
if isinstance(payload, dict):
|
|
return payload.get("workflows") or []
|
|
return []
|
|
|
|
def dispatch_workflow(self, workflow_id: str, ref: str = "main", inputs: dict | None = None) -> None:
|
|
"""
|
|
workflow_dispatch 이벤트 트리거
|
|
|
|
Args:
|
|
workflow_id: workflow 파일명 (e.g. 'ci.yml')
|
|
ref: 브랜치명 (기본 'main')
|
|
inputs: workflow_dispatch inputs dict
|
|
"""
|
|
url = f"{self._repo_url}/actions/workflows/{workflow_id}/dispatches"
|
|
body: dict[str, Any] = {"ref": ref}
|
|
if inputs:
|
|
body["inputs"] = inputs
|
|
self._request(url, method="POST", body=body)
|
|
|
|
# ------------------------------------------------------------------
|
|
# Pull Requests
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_prs(self, state: str = "open") -> list[dict]:
|
|
"""
|
|
PR 목록 조회
|
|
|
|
Args:
|
|
state: 'open' | 'closed' | 'all'
|
|
|
|
Returns:
|
|
PR 목록 (list[dict])
|
|
"""
|
|
url = f"{self._repo_url}/pulls?state={state}"
|
|
payload = self._request(url)
|
|
return payload if isinstance(payload, list) else []
|
|
|
|
def create_pr(
|
|
self,
|
|
title: str,
|
|
head: str,
|
|
base: str = "main",
|
|
body: str = "",
|
|
) -> dict:
|
|
"""
|
|
PR 생성
|
|
|
|
Args:
|
|
title: PR 제목
|
|
head: 소스 브랜치
|
|
base: 타겟 브랜치 (기본 'main')
|
|
body: PR 본문
|
|
|
|
Returns:
|
|
생성된 PR dict
|
|
"""
|
|
url = f"{self._repo_url}/pulls"
|
|
return self._request(url, method="POST", body={
|
|
"title": title,
|
|
"head": head,
|
|
"base": base,
|
|
"body": body,
|
|
})
|
|
|
|
# ------------------------------------------------------------------
|
|
# Releases
|
|
# ------------------------------------------------------------------
|
|
|
|
def get_latest_release(self) -> dict:
|
|
"""최신 릴리스 조회"""
|
|
url = f"{self._repo_url}/releases/latest"
|
|
return self._request(url)
|
|
|
|
def list_releases(self, limit: int = 10) -> list[dict]:
|
|
"""릴리스 목록 조회"""
|
|
url = f"{self._repo_url}/releases?limit={limit}"
|
|
payload = self._request(url)
|
|
return payload if isinstance(payload, list) else []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Secrets / Variables (관리용)
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_secrets(self) -> list[dict]:
|
|
"""저장소 secrets 이름 목록 (값은 반환되지 않음)"""
|
|
url = f"{self._repo_url}/actions/secrets"
|
|
payload = self._request(url)
|
|
return payload if isinstance(payload, list) else []
|
|
|
|
def list_variables(self) -> list[dict]:
|
|
"""저장소 actions variables 목록"""
|
|
url = f"{self._repo_url}/actions/variables"
|
|
payload = self._request(url)
|
|
if isinstance(payload, dict):
|
|
return payload.get("variables") or []
|
|
return []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Runners
|
|
# ------------------------------------------------------------------
|
|
|
|
def list_runners(self) -> list[dict]:
|
|
"""저장소에 연결된 Actions runner 목록"""
|
|
url = f"{self._api}/admin/runners"
|
|
payload = self._request(url)
|
|
if isinstance(payload, dict):
|
|
return payload.get("runners") or []
|
|
return []
|
|
|
|
# ------------------------------------------------------------------
|
|
# Utility
|
|
# ------------------------------------------------------------------
|
|
|
|
def summary(self) -> dict:
|
|
"""클라이언트 설정 요약 (토큰값 비노출)"""
|
|
return {
|
|
"base_url": self.base_url,
|
|
"owner": self.owner,
|
|
"repo": self.repo,
|
|
"token_source": self._token_source(),
|
|
"token_chars": len(self.token),
|
|
}
|
|
|
|
def _token_source(self) -> str:
|
|
"""어느 환경변수에서 토큰을 탐지했는지 반환"""
|
|
for key in _TOKEN_ENV_PRIORITY:
|
|
val = os.environ.get(key, "").strip()
|
|
if val and val == self.token:
|
|
return key
|
|
return "explicit_arg" if self.token else "none"
|