feat: add sovereign Gitea client tool
This commit is contained in:
59
tools/gitea_client.py
Normal file
59
tools/gitea_client.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""
|
||||
Gitea API Client — typed, sovereign, zero-dependency.
|
||||
|
||||
Enables the agent to interact with Timmy's sovereign Gitea instance
|
||||
for issue tracking, PR management, and knowledge persistence.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
from dataclasses import dataclass, field
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional, Dict, List
|
||||
|
||||
class GiteaClient:
|
||||
def __init__(self, base_url: Optional[str] = None, token: Optional[str] = None):
|
||||
self.base_url = base_url or os.environ.get("GITEA_URL", "http://143.198.27.163:3000")
|
||||
self.token = token or os.environ.get("GITEA_TOKEN")
|
||||
self.api = f"{self.base_url.rstrip('/')}/api/v1"
|
||||
|
||||
def _request(self, method: str, path: str, data: Optional[dict] = None) -> Any:
|
||||
url = f"{self.api}{path}"
|
||||
body = json.dumps(data).encode() if data else None
|
||||
req = urllib.request.Request(url, data=body, method=method)
|
||||
if self.token:
|
||||
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, timeout=30) as resp:
|
||||
raw = resp.read().decode()
|
||||
return json.loads(raw) if raw else {}
|
||||
except urllib.error.HTTPError as e:
|
||||
raise Exception(f"Gitea {e.code}: {e.read().decode()}") from e
|
||||
|
||||
def get_file(self, repo: str, path: str, ref: str = "main") -> Dict[str, Any]:
|
||||
return self._request("GET", f"/repos/{repo}/contents/{path}?ref={ref}")
|
||||
|
||||
def create_file(self, repo: str, path: str, content: str, message: str, branch: str = "main") -> Dict[str, Any]:
|
||||
data = {
|
||||
"branch": branch,
|
||||
"content": content, # Base64 encoded
|
||||
"message": message
|
||||
}
|
||||
return self._request("POST", f"/repos/{repo}/contents/{path}", data)
|
||||
|
||||
def update_file(self, repo: str, path: str, content: str, message: str, sha: str, branch: str = "main") -> Dict[str, Any]:
|
||||
data = {
|
||||
"branch": branch,
|
||||
"content": content, # Base64 encoded
|
||||
"message": message,
|
||||
"sha": sha
|
||||
}
|
||||
return self._request("PUT", f"/repos/{repo}/contents/{path}", data)
|
||||
Reference in New Issue
Block a user