115 lines
4.2 KiB
Python
115 lines
4.2 KiB
Python
"""
|
|
Forgejo API client and git operations for ifc-commit.
|
|
|
|
All network and git calls that would block the event loop are exposed as
|
|
synchronous functions meant to be run in a thread pool via
|
|
asyncio.run_in_executor().
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
import subprocess
|
|
import httpx
|
|
|
|
FORGE_BASE = "https://gitaec.org"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Forgejo API
|
|
# ---------------------------------------------------------------------------
|
|
|
|
async def list_yaml_files(repo: str, branch: str, token: str) -> list[str]:
|
|
"""
|
|
List yaml files inside the yaml/ directory of a Forgejo repository.
|
|
Returns filenames only (e.g. ["duplex.yaml", "office.yaml"]).
|
|
"""
|
|
url = f"{FORGE_BASE}/api/v1/repos/{repo}/contents/yaml"
|
|
headers = {"Authorization": f"token {token}", "Accept": "application/json"}
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(url, params={"ref": branch}, headers=headers)
|
|
r.raise_for_status()
|
|
entries = r.json()
|
|
return [
|
|
e["name"] for e in entries
|
|
if e.get("type") == "file" and e["name"].endswith((".yaml", ".yml"))
|
|
]
|
|
|
|
|
|
async def get_file_content(repo: str, path: str, branch: str, token: str) -> str:
|
|
"""
|
|
Fetch the raw content of a file from a Forgejo repository.
|
|
Uses the /contents/ API (returns base64) to avoid bot-detection on /raw/.
|
|
Raises httpx.HTTPStatusError on non-2xx responses.
|
|
"""
|
|
url = f"{FORGE_BASE}/api/v1/repos/{repo}/contents/{path}"
|
|
headers = {"Authorization": f"token {token}", "Accept": "application/json"}
|
|
async with httpx.AsyncClient() as client:
|
|
r = await client.get(url, params={"ref": branch}, headers=headers)
|
|
r.raise_for_status()
|
|
data = r.json()
|
|
return base64.b64decode(data["content"]).decode()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Git operations (synchronous — run in thread pool)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def clone_repo_shallow(repo: str, token: str, workdir: str) -> None:
|
|
"""Shallow clone (depth=1) — fast, no history, used for entity queries."""
|
|
url = f"https://oauth2:{token}@{FORGE_BASE.removeprefix('https://')}/{repo}.git"
|
|
_run(["git", "clone", "--depth=1", url, workdir])
|
|
|
|
|
|
def clone_repo(repo: str, token: str, workdir: str) -> None:
|
|
"""
|
|
Clone a Forgejo repository into workdir using token auth over HTTPS.
|
|
workdir must not exist yet (git clone creates it).
|
|
"""
|
|
url = f"https://oauth2:{token}@{FORGE_BASE.removeprefix('https://')}/{repo}.git"
|
|
_run(["git", "clone", url, workdir])
|
|
|
|
|
|
def commit_and_push(workdir: str, message: str, author_name: str = "ifc-commit", author_email: str = "ifc-commit@gitaec.org") -> bool:
|
|
"""
|
|
Stage all changes, commit, and push in workdir.
|
|
Returns True if a commit was made, False if there was nothing to commit.
|
|
Raises subprocess.CalledProcessError on real git failures.
|
|
"""
|
|
env = {
|
|
**os.environ,
|
|
"GIT_AUTHOR_NAME": author_name,
|
|
"GIT_AUTHOR_EMAIL": author_email,
|
|
"GIT_COMMITTER_NAME": author_name,
|
|
"GIT_COMMITTER_EMAIL": author_email,
|
|
}
|
|
_run(["git", "add", "-A"], cwd=workdir, env=env)
|
|
|
|
# Check if there is anything staged before committing
|
|
status = _run(["git", "status", "--porcelain"], cwd=workdir, env=env)
|
|
if not status.strip():
|
|
return False # nothing to commit
|
|
|
|
_run(["git", "commit", "-m", message], cwd=workdir, env=env)
|
|
_run(["git", "push"], cwd=workdir, env=env)
|
|
return True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Internal helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _run(cmd: list[str], cwd: str | None = None, env: dict | None = None) -> str:
|
|
"""Run a subprocess, raise on non-zero exit, return combined stdout+stderr."""
|
|
result = subprocess.run(
|
|
cmd,
|
|
cwd=cwd,
|
|
env=env,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
text=True,
|
|
)
|
|
if result.returncode != 0:
|
|
raise subprocess.CalledProcessError(
|
|
result.returncode, cmd, output=result.stdout
|
|
)
|
|
return result.stdout
|