m3taversal
9d69629893
Some checks failed
CI / lint-and-test (pull_request) Has been cancelled
- What: Unified forgejo_api(), get_pr_diff(), get_agent_token(), repo_path() into lib/forgejo.py. Removed 3 duplicate _forgejo_api functions (evaluate.py, merge.py, validate.py), 2 duplicate _get_pr_diff functions (evaluate.py, validate.py), and 1 _agent_token function (evaluate.py). - Why: Phase 3 structural refactor. Single source of truth for all Forgejo HTTP calls. Eliminates ~90 lines of duplicated code across 3 modules. - Connections: All hardcoded repo paths now use repo_path() helper. Consumer modules no longer reference config.FORGEJO_URL/OWNER/REPO/TOKEN_FILE directly. Pentagon-Agent: Ganymede <F99EBFA6-547B-4096-BEEA-1D59C3E4028A>
83 lines
3.1 KiB
Python
83 lines
3.1 KiB
Python
"""Forgejo API client — single shared module for all pipeline stages.
|
|
|
|
Extracted from evaluate.py, merge.py, validate.py (Phase 3 refactor).
|
|
All Forgejo HTTP calls go through this module.
|
|
"""
|
|
|
|
import logging
|
|
|
|
import aiohttp
|
|
|
|
from . import config
|
|
|
|
logger = logging.getLogger("pipeline.forgejo")
|
|
|
|
|
|
async def api(method: str, path: str, body: dict = None, token: str = None):
|
|
"""Call Forgejo API. Returns parsed JSON, {} for 204, or None on error.
|
|
|
|
Args:
|
|
method: HTTP method (GET, POST, DELETE, etc.)
|
|
path: API path after /api/v1 (e.g. "/repos/teleo/teleo-codex/pulls")
|
|
body: JSON body for POST/PUT/PATCH
|
|
token: Override token. If None, reads from FORGEJO_TOKEN_FILE (admin token).
|
|
"""
|
|
url = f"{config.FORGEJO_URL}/api/v1{path}"
|
|
if token is None:
|
|
token = config.FORGEJO_TOKEN_FILE.read_text().strip() if config.FORGEJO_TOKEN_FILE.exists() else ""
|
|
headers = {"Authorization": f"token {token}", "Content-Type": "application/json"}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.request(
|
|
method, url, headers=headers, json=body, timeout=aiohttp.ClientTimeout(total=60)
|
|
) as resp:
|
|
if resp.status >= 400:
|
|
text = await resp.text()
|
|
logger.error("Forgejo API %s %s → %d: %s", method, path, resp.status, text[:200])
|
|
return None
|
|
if resp.status == 204:
|
|
return {}
|
|
return await resp.json()
|
|
except Exception as e:
|
|
logger.error("Forgejo API error: %s %s → %s", method, path, e)
|
|
return None
|
|
|
|
|
|
async def get_pr_diff(pr_number: int) -> str:
|
|
"""Fetch PR diff via Forgejo API. Returns diff text or empty string."""
|
|
url = f"{config.FORGEJO_URL}/api/v1/repos/{config.FORGEJO_OWNER}/{config.FORGEJO_REPO}/pulls/{pr_number}.diff"
|
|
token = config.FORGEJO_TOKEN_FILE.read_text().strip() if config.FORGEJO_TOKEN_FILE.exists() else ""
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
url,
|
|
headers={"Authorization": f"token {token}", "Accept": "text/plain"},
|
|
timeout=aiohttp.ClientTimeout(total=60),
|
|
) as resp:
|
|
if resp.status >= 400:
|
|
return ""
|
|
diff = await resp.text()
|
|
if len(diff) > 2_000_000:
|
|
return ""
|
|
return diff
|
|
except Exception as e:
|
|
logger.error("Failed to fetch diff for PR #%d: %s", pr_number, e)
|
|
return ""
|
|
|
|
|
|
def get_agent_token(agent_name: str) -> str | None:
|
|
"""Read Forgejo token for a named agent. Returns token string or None."""
|
|
token_file = config.SECRETS_DIR / f"forgejo-{agent_name.lower()}-token"
|
|
if token_file.exists():
|
|
return token_file.read_text().strip()
|
|
return None
|
|
|
|
|
|
def repo_path(subpath: str = "") -> str:
|
|
"""Build standard repo API path: /repos/{owner}/{repo}/{subpath}."""
|
|
base = f"/repos/{config.FORGEJO_OWNER}/{config.FORGEJO_REPO}"
|
|
if subpath:
|
|
return f"{base}/{subpath}"
|
|
return base
|