From 02de35e1b0a8c396997c9100530e6bdcebf6cf81 Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Mon, 9 Jun 2025 18:27:06 +0200 Subject: [PATCH] Ruff after Claude Code --- aider_gitea/__init__.py | 8 +- aider_gitea/__main__.py | 5 +- aider_gitea/gitea_client.py | 248 ++++++++++++++++++--------- aider_gitea/models.py | 111 ++++++++++++ test/test_claude_code_integration.py | 14 +- test/test_gitea_client_pr_labels.py | 9 +- 6 files changed, 306 insertions(+), 89 deletions(-) create mode 100644 aider_gitea/models.py diff --git a/aider_gitea/__init__.py b/aider_gitea/__init__.py index 9fe3f23..ec0dc14 100644 --- a/aider_gitea/__init__.py +++ b/aider_gitea/__init__.py @@ -725,10 +725,10 @@ def solve_issues_in_repository( return for issue in issues: - issue_url = issue.get('web_url') - issue_number = issue.get('number') - issue_description = issue.get('body', '') - title = issue.get('title', f'Issue {issue_number}') + issue_url = issue.html_url + issue_number = str(issue.number) + issue_description = issue.body + title = issue.title if seen_issues_db.has_seen(issue_url): logger.info('Skipping already processed issue #%s: %s', issue_number, title) else: diff --git a/aider_gitea/__main__.py b/aider_gitea/__main__.py index e9ab35e..ee3a225 100644 --- a/aider_gitea/__main__.py +++ b/aider_gitea/__main__.py @@ -8,6 +8,8 @@ import argparse import logging import time +import requests + from . import RepositoryConfig, secrets, solve_issues_in_repository from .gitea_client import GiteaClient from .seen_issues_db import SeenIssuesDB @@ -69,7 +71,8 @@ def main(): core.EVALUATOR_MODEL = args.evaluator_model seen_issues_db = SeenIssuesDB() - client = GiteaClient(args.gitea_url, secrets.gitea_token()) + session = requests.Session() + client = GiteaClient(session, gitea_url=args.gitea_url, token=secrets.gitea_token()) if args.repo: repositories = [args.repo] diff --git a/aider_gitea/gitea_client.py b/aider_gitea/gitea_client.py index bf90768..0c0b379 100644 --- a/aider_gitea/gitea_client.py +++ b/aider_gitea/gitea_client.py @@ -1,10 +1,25 @@ import logging from collections.abc import Iterator +from typing import Any import requests +from .models import ( + GiteaIssue, + GiteaLabel, + GiteaUser, +) + logger = logging.getLogger(__name__) +# Module-level constants +API_VERSION_PATH = '/api/v1' +DEFAULT_CONTENT_TYPE = 'application/json' +AIDER_LABEL_NAME = 'aider' +SUCCESS_CONCLUSION = 'success' +CONFLICT_STATUS_CODE = 409 +UNPROCESSABLE_ENTITY_STATUS_CODE = 422 + class GiteaClient: """Client for interacting with the Gitea API. @@ -14,121 +29,191 @@ class GiteaClient: Read more about the Gitea API here: https://gitea.com/api/swagger + Follows the standardized client format: + 1. Constructor takes a requests.Session object + 2. All secrets are provided via keyword arguments + 3. ROOT_URL constant field for constructing URLs + Attributes: - gitea_url (str): The base URL for the Gitea API endpoints. session (requests.Session): HTTP session for making API requests. + ROOT_URL (str): The base URL for the Gitea API endpoints. """ - def __init__(self, gitea_url: str, token: str) -> None: + def __init__( + self, + session: requests.Session, + *, + gitea_url: str, + token: str = '', + ) -> None: """Initialize a new Gitea API client. Args: - gitea_url (str): Base URL for the Gitea instance (without '/api/v1'). - token (str): Authentication token for the Gitea API. If empty, requests will be unauthenticated. + session: HTTP session object to use for requests. + gitea_url: Base URL for the Gitea instance (without '/api/v1'). + token: Authentication token for the Gitea API. If empty, requests will be unauthenticated. Raises: AssertionError: If gitea_url ends with '/api/v1'. """ - assert not gitea_url.endswith('/api/v1') - self.gitea_url = gitea_url + '/api/v1' - self.session = requests.Session() - self.session.headers['Content-Type'] = 'application/json' + assert not gitea_url.endswith(API_VERSION_PATH) + self.session = session + self.ROOT_URL = gitea_url + API_VERSION_PATH + self.session.headers['Content-Type'] = DEFAULT_CONTENT_TYPE if token: self.session.headers['Authorization'] = f'token {token}' - def get_default_branch_sha(self, owner: str, repo: str, branch: str) -> str: + def get_default_branch_sha(self, owner: str, repo: str, branch_name: str) -> str: """Retrieve the commit SHA of the specified branch. Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. - branch (str): Name of the branch. + owner: Owner of the repository. + repo: Name of the repository. + branch_name: Name of the branch. Returns: - str: The commit SHA of the specified branch. + The commit SHA of the specified branch. Raises: requests.HTTPError: If the API request fails. """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}' - response = self.session.get(url) + api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/branches/{branch_name}' + response = self.session.get(api_url) response.raise_for_status() - data = response.json() - return data['commit']['sha'] + branch_data = response.json() + return branch_data['commit']['sha'] - def create_branch(self, owner: str, repo: str, new_branch: str, sha: str) -> bool: + def create_branch( + self, + owner: str, + repo: str, + new_branch_name: str, + commit_sha: str, + ) -> bool: """Create a new branch from the provided SHA. Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. - new_branch (str): Name of the new branch to create. - sha (str): Commit SHA to use as the starting point for the new branch. + owner: Owner of the repository. + repo: Name of the repository. + new_branch_name: Name of the new branch to create. + commit_sha: Commit SHA to use as the starting point for the new branch. Returns: - bool: True if the branch was created successfully, False if the branch already exists. + True if the branch was created successfully, False if the branch already exists. Raises: requests.HTTPError: If the API request fails for reasons other than branch already existing. """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/git/refs' - json_data = {'ref': f'refs/heads/{new_branch}', 'sha': sha} - response = self.session.post(url, json=json_data) - if response.status_code == 422: - logger.warning('Branch %s already exists.', new_branch) + api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/git/refs' + request_payload = {'ref': f'refs/heads/{new_branch_name}', 'sha': commit_sha} + response = self.session.post(api_url, json=request_payload) + if response.status_code == UNPROCESSABLE_ENTITY_STATUS_CODE: + logger.warning('Branch %s already exists.', new_branch_name) return False response.raise_for_status() return True - def get_issues(self, owner: str, repo: str) -> list[dict[str, str]]: + def get_issues(self, owner: str, repo: str) -> list[GiteaIssue]: """Download issues from the specified repository and filter those with the 'aider' label. Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. + owner: Owner of the repository. + repo: Name of the repository. Returns: - list: A list of issue dictionaries, filtered to only include issues with the 'aider' label. + A list of GiteaIssue objects, filtered to only include issues with the 'aider' label. Raises: requests.HTTPError: If the API request fails. """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/issues' - response = self.session.get(url) + api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/issues' + response = self.session.get(api_url) response.raise_for_status() - issues = response.json() + issues_data = response.json() + # Filter to only include issues marked with the "aider" label. - issues = [ - issue - for issue in issues - if any(label.get('name') == 'aider' for label in issue.get('labels', [])) + filtered_issues = [ + issue_data + for issue_data in issues_data + if any( + label_data.get('name') == AIDER_LABEL_NAME + for label_data in issue_data.get('labels', []) + ) ] - return issues + + # Convert to dataclass objects + gitea_issues = [] + for issue_data in filtered_issues: + labels = [ + GiteaLabel( + id=label_data['id'], + name=label_data['name'], + color=label_data['color'], + description=label_data.get('description', ''), + ) + for label_data in issue_data.get('labels', []) + ] + + user = GiteaUser( + login=issue_data['user']['login'], + id=issue_data['user']['id'], + full_name=issue_data['user'].get('full_name', ''), + email=issue_data['user'].get('email', ''), + avatar_url=issue_data['user'].get('avatar_url', ''), + ) + + assignees = [ + GiteaUser( + login=assignee_data['login'], + id=assignee_data['id'], + full_name=assignee_data.get('full_name', ''), + email=assignee_data.get('email', ''), + avatar_url=assignee_data.get('avatar_url', ''), + ) + for assignee_data in issue_data.get('assignees', []) + ] + + gitea_issue = GiteaIssue( + id=issue_data['id'], + number=issue_data['number'], + title=issue_data['title'], + body=issue_data.get('body', ''), + state=issue_data['state'], + labels=labels, + user=user, + assignees=assignees, + html_url=issue_data['html_url'], + created_at=issue_data['created_at'], + updated_at=issue_data['updated_at'], + ) + gitea_issues.append(gitea_issue) + + return gitea_issues def iter_user_repositories( self, - owner: str, + owner_name: str, only_those_with_issues: bool = False, ) -> Iterator[str]: """Get a list of repositories for a given user. Args: - owner (str): The owner of the repositories. - only_those_with_issues (bool): If True, only return repositories with issues enabled. + owner_name: The owner of the repositories. + only_those_with_issues: If True, only return repositories with issues enabled. Returns: - Iterator[str]: An iterator of repository names. + An iterator of repository names. """ - url = f'{self.gitea_url}/user/repos' - response = self.session.get(url) + api_url = f'{self.ROOT_URL}/user/repos' + response = self.session.get(api_url) response.raise_for_status() - for repo in response.json(): - if only_those_with_issues and not repo['has_issues']: + for repository_data in response.json(): + if only_those_with_issues and not repository_data['has_issues']: continue - if repo['owner']['login'].lower() != owner.lower(): + if repository_data['owner']['login'].lower() != owner_name.lower(): continue - yield repo['name'] + yield repository_data['name'] def create_pull_request( self, @@ -157,54 +242,61 @@ class GiteaClient: Raises: requests.HTTPError: If the API request fails. """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls' - json_data = { + api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls' + request_payload = { 'title': title, 'body': body, 'head': head, 'base': base, } - response = self.session.post(url, json=json_data) + response = self.session.post(api_url, json=request_payload) # If a pull request for this head/base already exists, return it instead of crashing - if response.status_code == 409: + if response.status_code == CONFLICT_STATUS_CODE: logger.warning( 'Pull request already exists for head %s and base %s', head, base, ) - prs = self.get_pull_requests(owner, repo) - for pr in prs: + existing_pull_requests = self.get_pull_requests(owner, repo) + for existing_pr in existing_pull_requests: if ( - pr.get('head', {}).get('ref') == head - and pr.get('base', {}).get('ref') == base + existing_pr.get('head', {}).get('ref') == head + and existing_pr.get('base', {}).get('ref') == base ): - return pr + return existing_pr # fallback to raise if we can’t find it response.raise_for_status() response.raise_for_status() return response.json() - def get_failed_pipelines(self, owner: str, repo: str, pr_number: str) -> list[int]: + def get_failed_pipelines( + self, + owner: str, + repo: str, + pull_request_number: str, + ) -> list[int]: """Fetch pipeline runs for a PR and return IDs of failed runs.""" - url = f'{self.gitea_url}/repos/{owner}/{repo}/actions/runs' - response = self.session.get(url) + api_url = f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs' + response = self.session.get(api_url) response.raise_for_status() - runs = response.json().get('workflow_runs', []) - failed = [] - for run in runs: + workflow_runs = response.json().get('workflow_runs', []) + failed_run_ids = [] + for workflow_run in workflow_runs: if any( - pr.get('number') == int(pr_number) - for pr in run.get('pull_requests', []) + pull_request.get('number') == int(pull_request_number) + for pull_request in workflow_run.get('pull_requests', []) ): - if run.get('conclusion') not in ('success',): - failed.append(run.get('id')) - return failed + if workflow_run.get('conclusion') != SUCCESS_CONCLUSION: + failed_run_ids.append(workflow_run.get('id')) + return failed_run_ids - def get_pipeline_log(self, owner: str, repo: str, run_id: int) -> str: + def get_pipeline_log(self, owner: str, repo: str, workflow_run_id: int) -> str: """Download the logs for a pipeline run.""" - url = f'{self.gitea_url}/repos/{owner}/{repo}/actions/runs/{run_id}/logs' - response = self.session.get(url) + api_url = ( + f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs/{workflow_run_id}/logs' + ) + response = self.session.get(api_url) response.raise_for_status() return response.text @@ -212,10 +304,12 @@ class GiteaClient: self, owner: str, repo: str, - state: str = 'open', - ) -> list[dict]: + pull_request_state: str = 'open', + ) -> list[dict[str, Any]]: """Fetch pull requests for a repository.""" - url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls?state={state}' - response = self.session.get(url) + api_url = ( + f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls?state={pull_request_state}' + ) + response = self.session.get(api_url) response.raise_for_status() return response.json() diff --git a/aider_gitea/models.py b/aider_gitea/models.py new file mode 100644 index 0000000..8214027 --- /dev/null +++ b/aider_gitea/models.py @@ -0,0 +1,111 @@ +"""Data models for Gitea API responses. + +This module contains dataclasses that represent the structure of data +returned from the Gitea API, providing type safety and better code organization. +""" + +from dataclasses import dataclass +from typing import Any + + +@dataclass +class GiteaUser: + """Represents a Gitea user.""" + + login: str # User's login name + id: int # User's ID + full_name: str # User's full name + email: str # User's email address + avatar_url: str # URL to user's avatar image + + +@dataclass +class GiteaLabel: + """Represents a label in Gitea.""" + + id: int # Label ID + name: str # Label name + color: str # Label color (hex code) + description: str # Label description + + +@dataclass +class GiteaCommit: + """Represents a commit in Gitea.""" + + sha: str # Commit SHA hash + url: str # API URL for the commit + message: str # Commit message + author: GiteaUser # Commit author + + +@dataclass +class GiteaBranch: + """Represents a branch in Gitea.""" + + name: str # Branch name + commit: GiteaCommit # Latest commit on the branch + protected: bool # Whether the branch is protected + + +@dataclass +class GiteaRepository: + """Represents a repository in Gitea.""" + + id: int # Repository ID + name: str # Repository name + full_name: str # Full repository name (owner/repo) + owner: GiteaUser # Repository owner + description: str # Repository description + clone_url: str # URL for cloning the repository + has_issues: bool # Whether issues are enabled + default_branch: str # Default branch name + + +@dataclass +class GiteaIssue: + """Represents an issue in Gitea.""" + + id: int # Issue ID + number: int # Issue number + title: str # Issue title + body: str # Issue description/body + state: str # Issue state (open, closed, etc.) + labels: list[GiteaLabel] # List of labels attached to the issue + user: GiteaUser # User who created the issue + assignees: list[GiteaUser] # List of assigned users + html_url: str # Web URL for the issue + created_at: str # ISO datetime when issue was created + updated_at: str # ISO datetime when issue was last updated + + +@dataclass +class GiteaPullRequest: + """Represents a pull request in Gitea.""" + + id: int # Pull request ID + number: int # Pull request number + title: str # Pull request title + body: str # Pull request description + state: str # Pull request state (open, closed, merged) + user: GiteaUser # User who created the pull request + head: dict[str, Any] # Head branch information + base: dict[str, Any] # Base branch information + html_url: str # Web URL for the pull request + created_at: str # ISO datetime when PR was created + updated_at: str # ISO datetime when PR was last updated + merged_at: str | None # ISO datetime when PR was merged (if applicable) + + +@dataclass +class GiteaWorkflowRun: + """Represents a GitHub Actions workflow run in Gitea.""" + + id: int # Workflow run ID + name: str # Workflow name + status: str # Run status (queued, in_progress, completed) + conclusion: str | None # Run conclusion (success, failure, cancelled, etc.) + workflow_id: int # ID of the workflow + pull_requests: list[dict[str, Any]] # Associated pull requests + created_at: str # ISO datetime when run was created + updated_at: str # ISO datetime when run was last updated diff --git a/test/test_claude_code_integration.py b/test/test_claude_code_integration.py index 9229225..ad83669 100644 --- a/test/test_claude_code_integration.py +++ b/test/test_claude_code_integration.py @@ -69,9 +69,10 @@ class TestClaudeCodeIntegration: 'claude', '-p', '--output-format', - 'json', - '--max-turns', - '10', + 'stream-json', + '--debug', + '--verbose', + '--dangerously-skip-permissions', issue, ] assert cmd == expected @@ -84,9 +85,10 @@ class TestClaudeCodeIntegration: 'claude', '-p', '--output-format', - 'json', - '--max-turns', - '10', + 'stream-json', + '--debug', + '--verbose', + '--dangerously-skip-permissions', '--model', 'claude-3-sonnet', issue, diff --git a/test/test_gitea_client_pr_labels.py b/test/test_gitea_client_pr_labels.py index 5277d0f..5ea3956 100644 --- a/test/test_gitea_client_pr_labels.py +++ b/test/test_gitea_client_pr_labels.py @@ -1,11 +1,18 @@ from unittest.mock import MagicMock, patch +import requests + from aider_gitea.gitea_client import GiteaClient class TestGiteaClientPRLabels: def setup_method(self): - self.client = GiteaClient('https://gitea.example.com', 'fake_token') + session = requests.Session() + self.client = GiteaClient( + session, + gitea_url='https://gitea.example.com', + token='fake_token', + ) @patch('requests.Session.post') def test_create_pull_request_with_labels(self, mock_post):