feat: editable guardrails, refs NOISSUE

This commit is contained in:
2026-04-11 11:12:50 +02:00
parent 798bb218f8
commit e35db0a361
9 changed files with 1559 additions and 103 deletions

View File

@@ -4,6 +4,20 @@ import os
import urllib.error
import urllib.request
import json
from urllib.parse import urlparse
def _normalize_base_url(base_url: str) -> str:
"""Normalize host-only service addresses into valid absolute URLs."""
normalized = (base_url or '').strip().rstrip('/')
if not normalized:
return ''
if '://' not in normalized:
normalized = f'https://{normalized}'
parsed = urlparse(normalized)
if not parsed.scheme or not parsed.netloc:
return ''
return normalized
class GiteaAPI:
@@ -11,7 +25,7 @@ class GiteaAPI:
def __init__(self, token: str, base_url: str, owner: str | None = None, repo: str | None = None):
self.token = token
self.base_url = base_url.rstrip("/")
self.base_url = _normalize_base_url(base_url)
self.owner = owner
self.repo = repo
self.headers = {
@@ -26,7 +40,7 @@ class GiteaAPI:
owner = os.getenv("GITEA_OWNER", "ai-test")
repo = os.getenv("GITEA_REPO", "")
return {
"base_url": base_url.rstrip("/"),
"base_url": _normalize_base_url(base_url),
"token": token,
"owner": owner,
"repo": repo,
@@ -96,16 +110,16 @@ class GiteaAPI:
def _request_sync(self, method: str, path: str, payload: dict | None = None) -> dict:
"""Perform a synchronous Gitea API request."""
request = urllib.request.Request(
self._api_url(path),
headers=self.get_auth_headers(),
method=method.upper(),
)
data = None
if payload is not None:
data = json.dumps(payload).encode('utf-8')
request.data = data
try:
if not self.base_url:
return {'error': 'Gitea base URL is not configured or is invalid'}
request = urllib.request.Request(
self._api_url(path),
headers=self.get_auth_headers(),
method=method.upper(),
)
if payload is not None:
request.data = json.dumps(payload).encode('utf-8')
with urllib.request.urlopen(request) as response:
body = response.read().decode('utf-8')
return json.loads(body) if body else {}
@@ -182,6 +196,10 @@ class GiteaAPI:
"""Get the user associated with the configured token."""
return await self._request("GET", "user")
def get_current_user_sync(self) -> dict:
"""Synchronously get the user associated with the configured token."""
return self._request_sync("GET", "user")
async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
"""Create a new branch."""
_owner = owner or self.owner