emil
d9ce4e240f
Post-merge / detect-type (push) Successful in 55s
Post-merge / release (push) Successful in 1m19s
Post-merge / validate-commit-msg (push) Successful in 1m19s
Post-merge / badges (push) Successful in 1m33s
Post-merge / vikunja (push) Successful in 1m13s
Post-merge / configure-repo (push) Successful in 1m13s
Post-merge / sync-wiki (push) Successful in 1m54s
Post-merge / publish (push) Successful in 55s
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""Minimal Gitea API client for workflow operations.
|
|
|
|
Uses urllib from the standard library to avoid adding requests as a
|
|
runtime dependency. Only covers the Actions workflow dispatch endpoint.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import urllib.error
|
|
import urllib.request # noqa: PTH123 # nosec B404
|
|
from contextlib import suppress
|
|
from typing import Any
|
|
|
|
|
|
class GiteaAPIError(Exception):
|
|
"""Raised when a Gitea API call fails."""
|
|
|
|
def __init__(self, status: int, message: str) -> None:
|
|
super().__init__(f"Gitea API error {status}: {message}")
|
|
self.status = status
|
|
self.message = message
|
|
|
|
|
|
class GiteaWorkflowClient:
|
|
"""Thin client for Gitea Actions workflow API endpoints."""
|
|
|
|
def __init__(self, base_url: str, token: str) -> None:
|
|
self._base_url = base_url.rstrip("/")
|
|
self._token = token
|
|
|
|
def _request(self, method: str, path: str, body: dict[str, Any] | None = None) -> dict[str, Any] | None:
|
|
url = f"{self._base_url}/api/v1{path}"
|
|
data = json.dumps(body).encode("utf-8") if body else None
|
|
req = urllib.request.Request( # nosec B310
|
|
url,
|
|
data=data,
|
|
method=method,
|
|
)
|
|
req.add_header("Authorization", f"token {self._token}")
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("Accept", "application/json")
|
|
try:
|
|
with urllib.request.urlopen(req) as resp: # noqa: PTH123 # nosec B310
|
|
if resp.status == 204:
|
|
return None
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else None
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8", errors="replace")
|
|
with suppress(json.JSONDecodeError, ValueError):
|
|
detail = json.loads(detail).get("message", detail)
|
|
raise GiteaAPIError(e.code, detail) from e
|
|
|
|
def list_workflows(self, owner: str, repo: str) -> list[dict[str, Any]]:
|
|
"""List all workflows in a repository."""
|
|
result = self._request("GET", f"/repos/{owner}/{repo}/actions/workflows")
|
|
if result is None:
|
|
return []
|
|
return result.get("workflows", [])
|
|
|
|
def dispatch_workflow(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
workflow_id: str,
|
|
ref: str = "master",
|
|
inputs: dict[str, str] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
"""Trigger a workflow dispatch event.
|
|
|
|
Args:
|
|
owner: Repository owner.
|
|
repo: Repository name.
|
|
workflow_id: Workflow file name (e.g. "ci.yml") or numeric ID.
|
|
ref: Git ref (branch/tag) to run on. Defaults to "master".
|
|
inputs: Optional workflow inputs.
|
|
|
|
Returns:
|
|
Run details dict if return_run_details is requested, else None.
|
|
"""
|
|
path = f"/repos/{owner}/{repo}/actions/workflows/{workflow_id}/dispatches?return_run_details=true"
|
|
body: dict[str, Any] = {"ref": ref}
|
|
if inputs:
|
|
body["inputs"] = inputs
|
|
return self._request("POST", path, body)
|