coulomb/markitect-main
2
0
Fork 0
Code Issues 61 Pull Requests Actions Projects 12 Releases 1 Wiki Activity

refactor: Factor out Gitea interfacing into clean facade pattern

Browse Source 
- Create new gitea/ package with clean API facade
- Establish proper separation of concerns: tddai uses gitea, not vice versa
- Replace duplicate curl+subprocess patterns with unified HTTP client
- Add rich domain models with properties (issue.priority, issue.status)
- Maintain full backwards compatibility in tddai modules
- Reduce code complexity: -373 lines, +151 lines (net -222 lines)
- Improve testability and maintainability through clean interfaces

Architecture:
- gitea.client.GiteaClient - main facade with sub-clients
- gitea.api_client - high-level API with model conversion
- gitea.http_client - low-level HTTP operations
- gitea.models - rich domain objects (Issue, Milestone, Label)
- gitea.config - gitea-specific configuration
- gitea.exceptions - clean exception hierarchy

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Bernd Worsch
2025-09-26 14:25:40 +02:00
parent b20b7003f5
commit fd8f792f08
11 changed files with 973 additions and 371 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
gitea/__init__.py Normal file
View File

@@ -0,0 +1,33 @@
"""
Gitea API facade - Clean interface for Gitea repository operations.
This package provides a clean, well-structured interface to Gitea API operations,
following the facade pattern to decouple application logic from specific API
implementation details.
Structure:
- client: Main GiteaClient facade
- models: Domain models (Issue, Milestone, Label, etc.)
- config: Gitea-specific configuration
- exceptions: Gitea-specific exceptions
Usage:
from gitea import GiteaClient
client = GiteaClient()
issues = client.issues.list()
issue = client.issues.get(42)
client.issues.create("Bug fix", "Description")
"""
from .client import GiteaClient
from .models import Issue, Milestone, Label, ProjectState, Priority
from .config import GiteaConfig
from .exceptions import GiteaError, GiteaAuthError, GiteaNotFoundError
__all__ = [
'GiteaClient',
'Issue', 'Milestone', 'Label', 'ProjectState', 'Priority',
'GiteaConfig',
'GiteaError', 'GiteaAuthError', 'GiteaNotFoundError'
]

203
gitea/api_client.py Normal file
View File

@@ -0,0 +1,203 @@
"""
High-level API client that converts between API responses and domain models.
"""
from datetime import datetime
from typing import List, Optional, Dict, Any
from .http_client import GiteaHttpClient
from .models import Issue, Milestone, Label, User, IssueCreateData, IssueUpdateData, MilestoneCreateData, LabelCreateData
from .config import GiteaConfig
from .exceptions import GiteaNotFoundError, GiteaError
class GiteaApiClient:
"""High-level API client with domain model conversion."""
def __init__(self, config: GiteaConfig):
self.config = config
self.http = GiteaHttpClient(config)
# Issue operations
def get_issue(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
try:
url = f"{self.config.issues_api_url}/{issue_number}"
data = self.http.get(url)
return self._parse_issue(data)
except GiteaError as e:
if "not found" in str(e).lower():
raise GiteaNotFoundError(f"Issue #{issue_number} not found")
raise
def list_issues(self, state: str = "all", page: int = 1, per_page: int = 50) -> List[Issue]:
"""List issues with optional filtering."""
params = {"page": str(page), "limit": str(per_page)}
if state != "all":
params["state"] = state
data = self.http.get(self.config.issues_api_url, params)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of issues")
return [self._parse_issue(issue_data) for issue_data in data]
def create_issue(self, issue_data: IssueCreateData) -> Issue:
"""Create a new issue."""
payload = {
"title": issue_data.title,
"body": issue_data.body,
}
if issue_data.assignees:
payload["assignees"] = issue_data.assignees
if issue_data.milestone:
payload["milestone"] = issue_data.milestone
if issue_data.labels:
payload["labels"] = issue_data.labels
data = self.http.post(self.config.issues_api_url, payload)
return self._parse_issue(data)
def update_issue(self, issue_number: int, update_data: IssueUpdateData) -> Issue:
"""Update an existing issue."""
payload = {}
if update_data.title is not None:
payload["title"] = update_data.title
if update_data.body is not None:
payload["body"] = update_data.body
if update_data.state is not None:
payload["state"] = update_data.state
if update_data.assignees is not None:
payload["assignees"] = update_data.assignees
if update_data.milestone is not None:
payload["milestone"] = update_data.milestone
if update_data.labels is not None:
payload["labels"] = update_data.labels
url = f"{self.config.issues_api_url}/{issue_number}"
data = self.http.patch(url, payload)
return self._parse_issue(data)
# Milestone operations
def list_milestones(self, state: str = "all") -> List[Milestone]:
"""List repository milestones."""
params = {}
if state != "all":
params["state"] = state
data = self.http.get(self.config.milestones_api_url, params)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of milestones")
return [self._parse_milestone(milestone_data) for milestone_data in data]
def create_milestone(self, milestone_data: MilestoneCreateData) -> Milestone:
"""Create a new milestone."""
payload = {
"title": milestone_data.title,
"description": milestone_data.description,
}
if milestone_data.due_on:
payload["due_on"] = milestone_data.due_on
data = self.http.post(self.config.milestones_api_url, payload)
return self._parse_milestone(data)
# Label operations
def list_labels(self) -> List[Label]:
"""List repository labels."""
data = self.http.get(self.config.labels_api_url)
if not isinstance(data, list):
raise GiteaError("Invalid response format: expected list of labels")
return [self._parse_label(label_data) for label_data in data]
def create_label(self, label_data: LabelCreateData) -> Label:
"""Create a new label."""
payload = {
"name": label_data.name,
"color": label_data.color,
"description": label_data.description,
}
data = self.http.post(self.config.labels_api_url, payload)
return self._parse_label(data)
# Parsing methods
def _parse_issue(self, data: Dict[str, Any]) -> Issue:
"""Parse issue data from API response."""
try:
# Parse labels
labels = []
if data.get('labels'):
labels = [self._parse_label(label_data) for label_data in data['labels']]
# Parse assignee
assignee = None
if data.get('assignee'):
assignee = self._parse_user(data['assignee'])
# Parse milestone
milestone = None
if data.get('milestone'):
milestone = self._parse_milestone(data['milestone'])
return Issue(
number=data['number'],
title=data['title'],
body=data.get('body', ''),
state=data['state'],
created_at=self._parse_datetime(data['created_at']),
updated_at=self._parse_datetime(data['updated_at']),
html_url=data['html_url'],
assignee=assignee,
labels=labels,
milestone=milestone
)
except (KeyError, ValueError) as e:
raise GiteaError(f"Failed to parse issue data: {e}")
def _parse_milestone(self, data: Dict[str, Any]) -> Milestone:
"""Parse milestone data from API response."""
return Milestone(
id=data['id'],
title=data['title'],
description=data.get('description', ''),
state=data['state'],
open_issues=data.get('open_issues', 0),
closed_issues=data.get('closed_issues', 0),
due_on=data.get('due_on'),
created_at=self._parse_datetime(data.get('created_at')) if data.get('created_at') else None,
updated_at=self._parse_datetime(data.get('updated_at')) if data.get('updated_at') else None
)
def _parse_label(self, data: Dict[str, Any]) -> Label:
"""Parse label data from API response."""
return Label(
id=data['id'],
name=data['name'],
color=data['color'],
description=data.get('description', '')
)
def _parse_user(self, data: Dict[str, Any]) -> User:
"""Parse user data from API response."""
return User(
id=data['id'],
login=data['login'],
full_name=data.get('full_name', ''),
email=data.get('email', ''),
avatar_url=data.get('avatar_url', '')
)
def _parse_datetime(self, date_str: str) -> datetime:
"""Parse datetime from API response."""
# Remove Z and microseconds for consistent parsing
date_str = date_str.replace('Z', '').split('.')[0]
return datetime.strptime(date_str, '%Y-%m-%dT%H:%M:%S')

195
gitea/client.py Normal file
View File

@@ -0,0 +1,195 @@
"""
Main Gitea client facade.
This provides a clean, organized interface for all Gitea operations,
following the facade pattern to hide complexity and provide a stable API.
"""
from typing import List, Optional
from .config import GiteaConfig
from .api_client import GiteaApiClient
from .models import Issue, Milestone, Label, IssueCreateData, IssueUpdateData, MilestoneCreateData, LabelCreateData, ProjectState, Priority
class IssuesClient:
"""Client for issue operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def get(self, issue_number: int) -> Issue:
"""Get a specific issue by number."""
return self._api.get_issue(issue_number)
def list(self, state: str = "all", page: int = 1, per_page: int = 50) -> List[Issue]:
"""List issues with optional filtering."""
return self._api.list_issues(state, page, per_page)
def list_open(self) -> List[Issue]:
"""List only open issues."""
return self._api.list_issues("open")
def list_closed(self) -> List[Issue]:
"""List only closed issues."""
return self._api.list_issues("closed")
def create(self, title: str, body: str = "", **kwargs) -> Issue:
"""Create a new issue."""
issue_data = IssueCreateData(
title=title,
body=body,
assignees=kwargs.get('assignees', []),
milestone=kwargs.get('milestone'),
labels=kwargs.get('labels', [])
)
return self._api.create_issue(issue_data)
def update(self, issue_number: int, **kwargs) -> Issue:
"""Update an existing issue."""
update_data = IssueUpdateData(
title=kwargs.get('title'),
body=kwargs.get('body'),
state=kwargs.get('state'),
assignees=kwargs.get('assignees'),
milestone=kwargs.get('milestone'),
labels=kwargs.get('labels')
)
return self._api.update_issue(issue_number, update_data)
def close(self, issue_number: int) -> Issue:
"""Close an issue."""
return self.update(issue_number, state="closed")
def reopen(self, issue_number: int) -> Issue:
"""Reopen an issue."""
return self.update(issue_number, state="open")
def add_labels(self, issue_number: int, labels: List[str]) -> Issue:
"""Add labels to an issue."""
issue = self.get(issue_number)
existing_labels = [label.name for label in issue.labels]
new_labels = list(set(existing_labels + labels))
return self.update(issue_number, labels=new_labels)
def remove_labels(self, issue_number: int, labels: List[str]) -> Issue:
"""Remove labels from an issue."""
issue = self.get(issue_number)
existing_labels = [label.name for label in issue.labels]
new_labels = [label for label in existing_labels if label not in labels]
return self.update(issue_number, labels=new_labels)
def set_priority(self, issue_number: int, priority: Priority) -> Issue:
"""Set issue priority."""
issue = self.get(issue_number)
labels = [label.name for label in issue.labels if not label.name.startswith('priority:')]
labels.append(priority.value)
return self.update(issue_number, labels=labels)
def set_status(self, issue_number: int, status: ProjectState) -> Issue:
"""Set issue status."""
issue = self.get(issue_number)
labels = [label.name for label in issue.labels if not label.name.startswith('status:')]
labels.append(status.value)
return self.update(issue_number, labels=labels)
class MilestonesClient:
"""Client for milestone operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def list(self, state: str = "all") -> List[Milestone]:
"""List milestones."""
return self._api.list_milestones(state)
def list_open(self) -> List[Milestone]:
"""List open milestones."""
return self._api.list_milestones("open")
def list_closed(self) -> List[Milestone]:
"""List closed milestones."""
return self._api.list_milestones("closed")
def create(self, title: str, description: str = "", due_on: str = None) -> Milestone:
"""Create a new milestone."""
milestone_data = MilestoneCreateData(
title=title,
description=description,
due_on=due_on
)
return self._api.create_milestone(milestone_data)
class LabelsClient:
"""Client for label operations."""
def __init__(self, api_client: GiteaApiClient):
self._api = api_client
def list(self) -> List[Label]:
"""List all labels."""
return self._api.list_labels()
def create(self, name: str, color: str, description: str = "") -> Label:
"""Create a new label."""
label_data = LabelCreateData(
name=name,
color=color,
description=description
)
return self._api.create_label(label_data)
def ensure_project_labels(self) -> None:
"""Ensure all standard project management labels exist."""
existing_labels = [label.name for label in self.list()]
# Define standard project labels
standard_labels = [
("status:todo", "d73a4a", "Ready to work on"),
("status:active", "0075ca", "Currently being worked on"),
("status:review", "fbca04", "Ready for review"),
("status:done", "0e8a16", "Completed work"),
("status:blocked", "b60205", "Blocked by dependencies"),
("priority:low", "c5def5", "Low priority"),
("priority:medium", "a2eeef", "Medium priority"),
("priority:high", "fef2c0", "High priority"),
("priority:critical", "d93f0b", "Critical priority"),
]
for name, color, description in standard_labels:
if name not in existing_labels:
self.create(name, color, description)
class GiteaClient:
"""Main Gitea client facade."""
def __init__(self, config: Optional[GiteaConfig] = None):
"""Initialize Gitea client.
Args:
config: GiteaConfig instance. If None, loads from environment.
"""
if config is None:
config = GiteaConfig.from_environment()
config.validate()
self.config = config
self._api = GiteaApiClient(config)
# Initialize sub-clients
self.issues = IssuesClient(self._api)
self.milestones = MilestonesClient(self._api)
self.labels = LabelsClient(self._api)
@classmethod
def from_tddai_config(cls, tddai_config) -> 'GiteaClient':
"""Create client from legacy TddaiConfig for backwards compatibility."""
gitea_config = GiteaConfig.from_tddai_config(tddai_config)
return cls(gitea_config)
def setup_project_management(self) -> None:
"""Setup standard project management labels and structure."""
self.labels.ensure_project_labels()

113
gitea/config.py Normal file
View File

@@ -0,0 +1,113 @@
"""
Gitea-specific configuration management.
"""
import os
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
from .exceptions import GiteaConfigError
def load_dotenv_file(env_file: Path) -> None:
"""Load environment variables from a .env file."""
if not env_file.exists():
return
with open(env_file, 'r') as f:
for line in f:
line = line.strip()
if line and not line.startswith('#') and '=' in line:
key, value = line.split('=', 1)
os.environ.setdefault(key.strip(), value.strip())
@dataclass
class GiteaConfig:
"""Configuration for Gitea API access."""
# Repository settings (required)
gitea_url: str = ""
repo_owner: str = ""
repo_name: str = ""
# Authentication (optional for read operations)
auth_token: Optional[str] = None
@property
def base_api_url(self) -> str:
"""Get the base API URL for this repository."""
return f"{self.gitea_url}/api/v1"
@property
def repo_api_url(self) -> str:
"""Get the repository API URL."""
return f"{self.base_api_url}/repos/{self.repo_owner}/{self.repo_name}"
@property
def issues_api_url(self) -> str:
"""Get the issues API URL."""
return f"{self.repo_api_url}/issues"
@property
def milestones_api_url(self) -> str:
"""Get the milestones API URL."""
return f"{self.repo_api_url}/milestones"
@property
def labels_api_url(self) -> str:
"""Get the labels API URL."""
return f"{self.repo_api_url}/labels"
@classmethod
def from_environment(cls, env_prefix: str = "GITEA") -> "GiteaConfig":
"""Create config from environment variables.
Args:
env_prefix: Environment variable prefix (default: GITEA)
Looks for {prefix}_URL, {prefix}_REPO_OWNER, etc.
"""
# Auto-load .env.gitea file if it exists
env_file = Path(".env.gitea")
load_dotenv_file(env_file)
config = cls()
# Load from environment
config.gitea_url = os.getenv(f"{env_prefix}_URL", "")
config.repo_owner = os.getenv(f"{env_prefix}_REPO_OWNER", "")
config.repo_name = os.getenv(f"{env_prefix}_REPO_NAME", "")
config.auth_token = os.getenv(f"{env_prefix}_API_TOKEN")
return config
@classmethod
def from_tddai_config(cls, tddai_config) -> "GiteaConfig":
"""Create GiteaConfig from legacy TddaiConfig for backwards compatibility."""
return cls(
gitea_url=tddai_config.gitea_url,
repo_owner=tddai_config.repo_owner,
repo_name=tddai_config.repo_name,
auth_token=os.getenv('GITEA_API_TOKEN')
)
def validate(self) -> None:
"""Validate configuration settings."""
if not self.gitea_url:
raise GiteaConfigError("gitea_url cannot be empty")
if not self.repo_owner:
raise GiteaConfigError("repo_owner cannot be empty")
if not self.repo_name:
raise GiteaConfigError("repo_name cannot be empty")
# Validate URL format
if not (self.gitea_url.startswith('http://') or self.gitea_url.startswith('https://')):
raise GiteaConfigError("gitea_url must start with http:// or https://")
def requires_auth(self, operation: str = "read") -> bool:
"""Check if operation requires authentication."""
write_operations = {"create", "update", "delete", "write"}
return operation in write_operations and not self.auth_token

31
gitea/exceptions.py Normal file
View File

@@ -0,0 +1,31 @@
"""
Gitea-specific exceptions.
"""
class GiteaError(Exception):
"""Base exception for Gitea API operations."""
pass
class GiteaAuthError(GiteaError):
"""Raised when authentication fails or token is missing."""
pass
class GiteaNotFoundError(GiteaError):
"""Raised when requested resource is not found."""
pass
class GiteaApiError(GiteaError):
"""Raised when API returns an error response."""
def __init__(self, message: str, status_code: int = None):
super().__init__(message)
self.status_code = status_code
class GiteaConfigError(GiteaError):
"""Raised when Gitea configuration is invalid or missing."""
pass

98
gitea/http_client.py Normal file
View File

@@ -0,0 +1,98 @@
"""
Low-level HTTP client for Gitea API operations.
This module handles the actual HTTP requests to Gitea API using subprocess + curl
for maximum compatibility and minimal dependencies.
"""
import json
import subprocess
from subprocess import PIPE
from typing import Dict, Any, Optional, List
from .exceptions import GiteaError, GiteaApiError, GiteaAuthError
from .config import GiteaConfig
class GiteaHttpClient:
"""Low-level HTTP client for Gitea API."""
def __init__(self, config: GiteaConfig):
self.config = config
def get(self, url: str, params: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
"""Make GET request to Gitea API."""
if params:
param_string = '&'.join(f"{k}={v}" for k, v in params.items())
url = f"{url}?{param_string}"
return self._make_request('GET', url)
def post(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make POST request to Gitea API."""
self._require_auth()
return self._make_request('POST', url, data)
def patch(self, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make PATCH request to Gitea API."""
self._require_auth()
return self._make_request('PATCH', url, data)
def delete(self, url: str) -> Dict[str, Any]:
"""Make DELETE request to Gitea API."""
self._require_auth()
return self._make_request('DELETE', url)
def _make_request(self, method: str, url: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Make HTTP request using curl."""
cmd = ['curl', '-s', '-X', method]
# Add authentication if available
if self.config.auth_token:
cmd.extend(['-H', f'Authorization: token {self.config.auth_token}'])
# Add content type for requests with data
if data is not None:
cmd.extend(['-H', 'Content-Type: application/json'])
cmd.extend(['-d', json.dumps(data)])
cmd.append(url)
try:
result = subprocess.run(
cmd,
stdout=PIPE,
stderr=PIPE,
universal_newlines=True,
check=True
)
if result.returncode != 0:
raise GiteaApiError(f"HTTP request failed: {result.stderr}")
# Handle empty responses
if not result.stdout.strip():
return {}
response_data = json.loads(result.stdout)
# Check for API error responses
if isinstance(response_data, dict):
if 'message' in response_data:
# This could be an error or just a response with a message field
# We need to distinguish based on context or HTTP status
if any(error_word in response_data['message'].lower()
for error_word in ['error', 'not found', 'forbidden', 'unauthorized']):
raise GiteaApiError(response_data['message'])
return response_data
except subprocess.CalledProcessError as e:
raise GiteaApiError(f"HTTP request failed: {e.stderr}")
except json.JSONDecodeError as e:
raise GiteaError(f"Failed to parse API response: {e}")
def _require_auth(self):
"""Ensure authentication token is available."""
if not self.config.auth_token:
raise GiteaAuthError("Authentication token required for this operation")

151
gitea/models.py Normal file
View File

@@ -0,0 +1,151 @@
"""
Gitea domain models.
These models represent the core entities in Gitea and provide a clean interface
independent of the underlying API representation.
"""
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from typing import List, Optional, Dict, Any
class ProjectState(Enum):
"""Standard project states using labels."""
TODO = "status:todo"
ACTIVE = "status:active"
REVIEW = "status:review"
DONE = "status:done"
BLOCKED = "status:blocked"
class Priority(Enum):
"""Priority levels using labels."""
LOW = "priority:low"
MEDIUM = "priority:medium"
HIGH = "priority:high"
CRITICAL = "priority:critical"
@dataclass
class Label:
"""Represents a Gitea issue label."""
id: int
name: str
color: str
description: str = ""
@dataclass
class User:
"""Represents a Gitea user."""
id: int
login: str
full_name: str = ""
email: str = ""
avatar_url: str = ""
@dataclass
class Milestone:
"""Represents a Gitea milestone (used as projects)."""
id: int
title: str
description: str
state: str # 'open' or 'closed'
open_issues: int
closed_issues: int
due_on: Optional[str] = None
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
@dataclass
class Issue:
"""Represents a Gitea issue."""
number: int
title: str
body: str
state: str # 'open' or 'closed'
created_at: datetime
updated_at: datetime
html_url: str
assignee: Optional[User] = None
labels: List[Label] = None
milestone: Optional[Milestone] = None
def __post_init__(self):
if self.labels is None:
self.labels = []
@property
def priority(self) -> Optional[str]:
"""Get issue priority from labels."""
for label in self.labels:
if label.name.startswith('priority:'):
return label.name.replace('priority:', '')
return None
@property
def status(self) -> Optional[str]:
"""Get issue status from labels."""
for label in self.labels:
if label.name.startswith('status:'):
return label.name.replace('status:', '')
return None
def has_label(self, label_name: str) -> bool:
"""Check if issue has a specific label."""
return any(label.name == label_name for label in self.labels)
def has_priority(self, priority: Priority) -> bool:
"""Check if issue has a specific priority."""
return self.has_label(priority.value)
def has_status(self, status: ProjectState) -> bool:
"""Check if issue has a specific status."""
return self.has_label(status.value)
@dataclass
class IssueCreateData:
"""Data for creating a new issue."""
title: str
body: str = ""
assignees: List[str] = None
milestone: Optional[int] = None
labels: List[str] = None
def __post_init__(self):
if self.assignees is None:
self.assignees = []
if self.labels is None:
self.labels = []
@dataclass
class IssueUpdateData:
"""Data for updating an existing issue."""
title: Optional[str] = None
body: Optional[str] = None
state: Optional[str] = None
assignees: Optional[List[str]] = None
milestone: Optional[int] = None
labels: Optional[List[str]] = None
@dataclass
class MilestoneCreateData:
"""Data for creating a new milestone."""
title: str
description: str = ""
due_on: Optional[str] = None
@dataclass
class LabelCreateData:
"""Data for creating a new label."""
name: str
color: str
description: str = ""