feat(gitea-harness): add tools/gitea/ package - GiteaClient + harness CLI with GITEA_TOKEN_TAXBAIK auto-detection [WBS-10]
- 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
This commit is contained in:
@@ -0,0 +1,315 @@
|
||||
"""
|
||||
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"
|
||||
Reference in New Issue
Block a user