464 lines
17 KiB
Python
464 lines
17 KiB
Python
"""Reusable HTTP API clients for Gitea and Vikunja."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
from typing import Any
|
|
|
|
import requests
|
|
from tenacity import (
|
|
before_sleep_log,
|
|
retry,
|
|
retry_if_exception_type,
|
|
stop_after_attempt,
|
|
wait_exponential,
|
|
)
|
|
|
|
from devx.config import DEFAULT_TIMEOUT, MAX_RETRIES, RETRY_BACKOFF_BASE, RETRY_STATUS_CODES
|
|
from devx.exceptions import APIError
|
|
|
|
logger = logging.getLogger("devx")
|
|
|
|
|
|
def _parse_error(e: requests.HTTPError) -> tuple[int, str]:
|
|
"""Extract status code and message from an HTTPError response."""
|
|
response = getattr(e, "response", None)
|
|
status = response.status_code if response is not None else 0
|
|
try:
|
|
body: dict[str, Any] = response.json() if response is not None else {}
|
|
message: str = body.get("message", str(e))
|
|
except (json.JSONDecodeError, ValueError, AttributeError):
|
|
message = str(e)
|
|
return status, message
|
|
|
|
|
|
class _TransientHTTPError(requests.HTTPError):
|
|
"""HTTP error with a retryable status code (wrapped for tenacity)."""
|
|
|
|
|
|
class _RetryableRequestError(Exception):
|
|
"""Connection/timeout error wrapped for tenacity retry."""
|
|
|
|
|
|
def _execute_request(
|
|
session: requests.Session,
|
|
method: str,
|
|
url: str,
|
|
**kwargs: Any,
|
|
) -> requests.Response:
|
|
"""Execute a single HTTP request, wrapping transient errors for tenacity.
|
|
|
|
Non-retryable HTTP errors (4xx except 429) raise :class:`APIError` directly.
|
|
Retryable errors (429, 5xx, connection, timeout) raise exceptions that
|
|
tenacity will retry.
|
|
"""
|
|
try:
|
|
response = session.request(method, url, timeout=DEFAULT_TIMEOUT, **kwargs)
|
|
response.raise_for_status()
|
|
return response
|
|
except requests.HTTPError as e:
|
|
status, message = _parse_error(e)
|
|
if status in RETRY_STATUS_CODES:
|
|
# Wrap in _TransientHTTPError so tenacity retries it
|
|
raise _TransientHTTPError(message, response=e.response) from e
|
|
raise APIError(status, message) from e
|
|
except (requests.ConnectionError, requests.Timeout) as e:
|
|
raise _RetryableRequestError(str(e)) from e
|
|
|
|
|
|
# Tenacity retry decorator shared by both clients.
|
|
# Retries on transient HTTP errors (429, 5xx) and connection/timeout errors.
|
|
_retry_decorator = retry(
|
|
stop=stop_after_attempt(MAX_RETRIES),
|
|
wait=wait_exponential(multiplier=RETRY_BACKOFF_BASE, min=RETRY_BACKOFF_BASE, max=RETRY_BACKOFF_BASE**MAX_RETRIES),
|
|
retry=retry_if_exception_type((_TransientHTTPError, _RetryableRequestError)),
|
|
before_sleep=before_sleep_log(logger, logging.WARNING),
|
|
reraise=True,
|
|
)
|
|
|
|
|
|
def _request_with_retry(
|
|
session: requests.Session,
|
|
url: str,
|
|
method: str,
|
|
**kwargs: Any,
|
|
) -> requests.Response:
|
|
"""Execute an HTTP request with tenacity-managed retry logic.
|
|
|
|
On exhaustion, the last exception is translated to :class:`APIError`.
|
|
"""
|
|
try:
|
|
return _retry_decorator(_execute_request)(session, method, url, **kwargs)
|
|
except _TransientHTTPError as e:
|
|
response = getattr(e, "response", None)
|
|
status = response.status_code if response is not None else 0
|
|
raise APIError(status, str(e)) from e
|
|
except _RetryableRequestError as e:
|
|
raise APIError(0, str(e)) from e
|
|
|
|
|
|
class GiteaClient:
|
|
"""Low-level Gitea REST API client with connection pooling."""
|
|
|
|
def __init__(self, base_url: str, token: str, owner: str, repo: str) -> None:
|
|
self._base_url = base_url.rstrip("/")
|
|
self._owner = owner
|
|
self._repo = repo
|
|
self._session = requests.Session()
|
|
self._session.headers.update(
|
|
{
|
|
"Authorization": f"token {token}",
|
|
"Content-Type": "application/json",
|
|
}
|
|
)
|
|
|
|
def _url(self, path: str) -> str:
|
|
return f"{self._base_url}/repos/{self._owner}/{self._repo}{path}"
|
|
|
|
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
|
|
return _request_with_retry(self._session, self._url(path), method, **kwargs)
|
|
|
|
# -- repo settings --
|
|
|
|
def update_repo_settings(self, settings: dict[str, Any]) -> dict[str, Any]:
|
|
"""Update repository settings (e.g. auto-delete branch after merge)."""
|
|
r = self._request("PATCH", "", json=settings)
|
|
return r.json()
|
|
|
|
# -- branch protection --
|
|
|
|
def list_branch_protections(self) -> list[dict[str, Any]]:
|
|
r = self._request("GET", "/branch_protections")
|
|
return r.json()
|
|
|
|
def create_branch_protection(self, config: dict[str, Any]) -> dict[str, Any]:
|
|
r = self._request("POST", "/branch_protections", json=config)
|
|
return r.json()
|
|
|
|
def update_branch_protection(self, branch: str, config: dict[str, Any]) -> dict[str, Any]:
|
|
r = self._request("PATCH", f"/branch_protections/{branch}", json=config)
|
|
return r.json()
|
|
|
|
def ensure_branch_protection(self, branch: str, config: dict[str, Any]) -> dict[str, Any]:
|
|
"""Idempotent: create or update branch protection for the given branch."""
|
|
existing = self.list_branch_protections()
|
|
for p in existing:
|
|
if p.get("branch_name") == branch:
|
|
update_config = {k: v for k, v in config.items() if k != "branch_name"}
|
|
return self.update_branch_protection(branch, update_config)
|
|
return self.create_branch_protection(config)
|
|
|
|
# -- labels --
|
|
|
|
def list_labels(self) -> list[dict[str, Any]]:
|
|
r = self._request("GET", "/labels")
|
|
return r.json()
|
|
|
|
def create_label(self, name: str, color: str, description: str = "") -> dict[str, Any]:
|
|
r = self._request(
|
|
"POST",
|
|
"/labels",
|
|
json={"name": name, "color": color, "description": description},
|
|
)
|
|
return r.json()
|
|
|
|
def ensure_label(self, name: str, color: str, description: str = "") -> dict[str, Any] | None:
|
|
"""Idempotent: create label if it doesn't already exist."""
|
|
labels = self.list_labels()
|
|
for label in labels:
|
|
if label["name"] == name:
|
|
return None
|
|
return self.create_label(name, color, description)
|
|
|
|
def create_issue(self, title: str, body: str = "", labels: list[int] | None = None) -> dict[str, Any]:
|
|
"""Create a new issue in the repository.
|
|
|
|
Args:
|
|
labels: List of label IDs (integers, not names).
|
|
"""
|
|
payload: dict[str, Any] = {"title": title, "body": body}
|
|
if labels:
|
|
payload["labels"] = labels
|
|
r = self._request("POST", "/issues", json=payload)
|
|
return r.json()
|
|
|
|
# -- pulls / releases --
|
|
|
|
def get_pr_labels(self, pr_number: str | int) -> list[dict[str, Any]]:
|
|
"""Fetch labels currently attached to a pull request."""
|
|
r = self._request("GET", f"/issues/{pr_number}/labels")
|
|
return r.json()
|
|
|
|
def merge_pr(self, pr_number: str | int, merge_title: str) -> None:
|
|
payload = {"Do": "squash", "MergeTitleField": merge_title}
|
|
self._request("POST", f"/pulls/{pr_number}/merge", json=payload)
|
|
|
|
def update_pr_branch(self, pr_number: str | int, style: str = "rebase") -> None:
|
|
"""Update PR head branch by merging/rebasing the base branch into it.
|
|
|
|
Uses the Gitea API ``POST /pulls/{index}/update?style=rebase`` endpoint.
|
|
This rebases the PR's head branch onto the latest base branch server-side,
|
|
triggering a ``pull_request synchronize`` event that starts a new CI run.
|
|
|
|
Args:
|
|
pr_number: PR number.
|
|
style: Update method — ``"rebase"`` (default) or ``"merge"``.
|
|
"""
|
|
self._request("POST", f"/pulls/{pr_number}/update", params={"style": style})
|
|
|
|
def get_commit_status(self, sha: str) -> list[dict[str, Any]]:
|
|
"""Fetch all status check contexts reported for a commit.
|
|
|
|
Uses the combined status endpoint (/commits/{sha}/status) which
|
|
returns one entry per context (the latest), deduplicated server-side.
|
|
The plural endpoint (/commits/{sha}/statuses) returns every historical
|
|
entry including stale "pending" ones that never got updated.
|
|
"""
|
|
r = self._request("GET", f"/commits/{sha}/status")
|
|
data = r.json()
|
|
return data.get("statuses", [])
|
|
|
|
def get_pr(self, pr_number: str | int) -> dict[str, Any]:
|
|
"""Fetch pull request details including mergeable state."""
|
|
r = self._request("GET", f"/pulls/{pr_number}")
|
|
return r.json()
|
|
|
|
def create_pr(self, title: str, head: str, base: str = "master", body: str = "") -> dict[str, Any]:
|
|
"""Create a pull request and return the PR dict.
|
|
|
|
Args:
|
|
title: PR title.
|
|
head: Head branch name.
|
|
base: Base branch name (default: master).
|
|
body: PR description (markdown).
|
|
"""
|
|
payload: dict[str, Any] = {"title": title, "head": head, "base": base}
|
|
if body:
|
|
payload["body"] = body
|
|
r = self._request("POST", "/pulls", json=payload)
|
|
return r.json()
|
|
|
|
def list_prs(self, state: str = "all", **params: Any) -> list[dict[str, Any]]:
|
|
"""List pull requests, optionally filtered by state.
|
|
|
|
Args:
|
|
state: ``open``, ``closed``, ``all`` (default).
|
|
**params: Additional query params (e.g. ``q="keyword"`` for title search).
|
|
"""
|
|
params.setdefault("state", state)
|
|
r = self._request("GET", "/pulls", params=params)
|
|
return r.json()
|
|
|
|
def get_pr_files(self, pr_number: str | int) -> list[dict[str, Any]]:
|
|
"""Fetch the list of files changed in a pull request."""
|
|
r = self._request("GET", f"/pulls/{pr_number}/files")
|
|
return r.json()
|
|
|
|
def add_pr_label(self, pr_number: str | int, label_names: list[str]) -> None:
|
|
"""Attach labels to a PR/issue by name.
|
|
|
|
Args:
|
|
pr_number: PR or issue number.
|
|
label_names: List of label names to attach.
|
|
"""
|
|
self._request("POST", f"/issues/{pr_number}/labels", json={"labels": label_names})
|
|
|
|
def get_pr_label_names(self, pr_number: str | int) -> list[str]:
|
|
"""Return label names currently attached to a PR/issue."""
|
|
r = self._request("GET", f"/issues/{pr_number}/labels")
|
|
return [label.get("name", "") for label in r.json()]
|
|
|
|
def get_pr_commits(self, pr_number: str | int) -> list[dict[str, Any]]:
|
|
"""Fetch the commits included in a pull request."""
|
|
r = self._request("GET", f"/pulls/{pr_number}/commits")
|
|
return r.json()
|
|
|
|
def get_pr_reviews(self, pr_number: str | int) -> list[dict[str, Any]]:
|
|
"""Fetch reviews posted on a pull request."""
|
|
r = self._request("GET", f"/pulls/{pr_number}/reviews")
|
|
return r.json()
|
|
|
|
def create_review(
|
|
self,
|
|
pr_number: str | int,
|
|
event: str = "COMMENT",
|
|
body: str = "",
|
|
comments: list[dict[str, Any]] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Post a review on a pull request.
|
|
|
|
Args:
|
|
event: ``APPROVED``, ``REQUEST_CHANGES``, or ``COMMENT``.
|
|
body: Top-level review body text.
|
|
comments: Line-level comments with ``path``, ``body``,
|
|
``new_position`` (and optionally ``old_position``).
|
|
"""
|
|
# Map common event names to Gitea API values
|
|
event_map = {"APPROVE": "APPROVED", "REQUEST_CHANGES": "REQUEST_CHANGES", "COMMENT": "COMMENT"}
|
|
gitea_event = event_map.get(event, event)
|
|
payload: dict[str, Any] = {"event": gitea_event, "body": body}
|
|
if comments:
|
|
payload["comments"] = comments
|
|
r = self._request("POST", f"/pulls/{pr_number}/reviews", json=payload)
|
|
return r.json()
|
|
|
|
def create_release(
|
|
self,
|
|
tag: str,
|
|
name: str = "",
|
|
body: str = "",
|
|
draft: bool = False,
|
|
prerelease: bool = False,
|
|
) -> dict[str, Any]:
|
|
payload = {
|
|
"tag_name": tag,
|
|
"name": name or tag,
|
|
"body": body,
|
|
"draft": draft,
|
|
"prerelease": prerelease,
|
|
}
|
|
r = self._request("POST", "/releases", json=payload)
|
|
return r.json()
|
|
|
|
def get_release_by_tag(self, tag: str) -> dict[str, Any] | None:
|
|
"""Fetch a release by its tag name. Returns None if not found."""
|
|
try:
|
|
r = self._request("GET", f"/releases/tags/{tag}")
|
|
return r.json()
|
|
except APIError:
|
|
return None
|
|
|
|
def create_release_idempotent(
|
|
self,
|
|
tag: str,
|
|
name: str = "",
|
|
body: str = "",
|
|
draft: bool = False,
|
|
prerelease: bool = False,
|
|
) -> dict[str, Any]:
|
|
"""Create a release, or return the existing one if it already exists.
|
|
|
|
This is idempotent — safe to call multiple times for the same tag.
|
|
"""
|
|
existing = self.get_release_by_tag(tag)
|
|
if existing:
|
|
logger.info("Release for tag %s already exists (ID %s), skipping creation.", tag, existing.get("id"))
|
|
return existing
|
|
return self.create_release(tag=tag, name=name, body=body, draft=draft, prerelease=prerelease)
|
|
|
|
# -- actions (CI/CD) --
|
|
|
|
def list_action_runs(self, **params: Any) -> dict[str, Any]:
|
|
"""List workflow runs for the repository.
|
|
|
|
Returns the raw API response dict (includes ``workflow_runs`` and
|
|
``total_count`` keys per Gitea API).
|
|
"""
|
|
r = self._request("GET", "/actions/runs", params=params)
|
|
return r.json()
|
|
|
|
def get_action_run_jobs(self, run_id: str | int) -> list[dict[str, Any]]:
|
|
"""List jobs for a specific workflow run."""
|
|
r = self._request("GET", f"/actions/runs/{run_id}/jobs")
|
|
data = r.json()
|
|
return data.get("jobs", [])
|
|
|
|
def get_action_job_logs(self, job_id: str | int) -> str:
|
|
"""Fetch logs for a specific CI job.
|
|
|
|
Returns the raw log text. Raises APIError if logs are unavailable.
|
|
"""
|
|
r = self._request("GET", f"/actions/jobs/{job_id}/logs")
|
|
return r.text
|
|
|
|
|
|
class VikunjaClient:
|
|
"""Low-level Vikunja REST API client with connection pooling."""
|
|
|
|
def __init__(self, base_url: str, token: str) -> None:
|
|
self._base_url = base_url.rstrip("/")
|
|
self._session = requests.Session()
|
|
self._session.headers.update({"Authorization": f"Bearer {token}"})
|
|
|
|
def _request(self, method: str, path: str, **kwargs: Any) -> requests.Response:
|
|
url = f"{self._base_url}{path}"
|
|
return _request_with_retry(self._session, url, method, **kwargs)
|
|
|
|
def list_tasks(self, **params: Any) -> list[dict[str, Any]]:
|
|
r = self._request("GET", "/tasks", params=params)
|
|
return r.json()
|
|
|
|
def get_task(self, task_id: int) -> dict[str, Any]:
|
|
"""Fetch a single task by its numeric ID."""
|
|
r = self._request("GET", f"/tasks/{task_id}")
|
|
return r.json()
|
|
|
|
def list_project_tasks(self, project_id: int, **params: Any) -> list[dict[str, Any]]:
|
|
"""List tasks in a specific project (more efficient than listing all tasks)."""
|
|
r = self._request("GET", f"/projects/{project_id}/tasks", params=params)
|
|
return r.json()
|
|
|
|
def find_task_by_identifier(self, project_id: int, identifier: str, per_page: int = 50) -> dict[str, Any] | None:
|
|
"""Find a task by its identifier (e.g. ``DEVX-42``) in a project.
|
|
|
|
Paginates through all tasks in the project. Returns the task dict
|
|
or None if not found.
|
|
"""
|
|
page = 1
|
|
while True:
|
|
tasks = self.list_project_tasks(project_id, page=page, per_page=per_page)
|
|
if not tasks:
|
|
break
|
|
for t in tasks:
|
|
if t.get("identifier") == identifier:
|
|
return t
|
|
if len(tasks) < per_page:
|
|
break
|
|
page += 1
|
|
return None
|
|
|
|
def create_task(self, project_id: int, title: str, description: str = "") -> dict[str, Any]:
|
|
"""Create a task in a project and return the created task dict.
|
|
|
|
Args:
|
|
project_id: Target Vikunja project ID.
|
|
title: Task title (required, non-empty).
|
|
description: Task description (HTML supported, optional).
|
|
"""
|
|
r = self._request(
|
|
"PUT",
|
|
f"/projects/{project_id}/tasks",
|
|
json={"title": title, "description": description},
|
|
)
|
|
return r.json()
|
|
|
|
def post_comment(self, task_id: int, comment: str) -> None:
|
|
self._request("PUT", f"/tasks/{task_id}/comments", json={"comment": comment})
|
|
|
|
def list_comments(self, task_id: int) -> list[dict[str, Any]]:
|
|
"""List all comments on a task."""
|
|
r = self._request("GET", f"/tasks/{task_id}/comments")
|
|
return r.json()
|
|
|
|
def update_task(self, task_id: int, **fields: Any) -> None:
|
|
"""Update task fields via POST (full replacement semantics).
|
|
|
|
Warning: Vikunja's POST /tasks/{id} replaces the entire task body.
|
|
Unspecified fields are reset to their type defaults. Use
|
|
``update_task_safe`` to preserve existing fields.
|
|
"""
|
|
self._request("POST", f"/tasks/{task_id}", json=fields)
|
|
|
|
def update_task_safe(self, task_id: int, **fields: Any) -> dict[str, Any]:
|
|
"""Safely update task fields using read-merge-write pattern.
|
|
|
|
Fetches the full task body, merges the provided fields on top,
|
|
and POSTs the complete body back. This prevents accidental
|
|
resets of done status, title, etc.
|
|
"""
|
|
task = self.get_task(task_id)
|
|
task.update(fields)
|
|
r = self._request("POST", f"/tasks/{task_id}", json=task)
|
|
return r.json()
|