From 69732c77b0b690b25c6978670e6d772f7f59e92a Mon Sep 17 00:00:00 2001
From: Jon Michael Aanes <jonjmaa@gmail.com>
Date: Mon, 9 Jun 2025 18:36:25 +0200
Subject: [PATCH] Ruff after Claude Code
---
aider_gitea/__init__.py | 8 +--
aider_gitea/gitea_client.py | 127 +++++++++++++++++++++++++++++++++---
2 files changed, 120 insertions(+), 15 deletions(-)
diff --git a/aider_gitea/__init__.py b/aider_gitea/__init__.py
index 9fe3f23..9d907d8 100644
--- a/aider_gitea/__init__.py
+++ b/aider_gitea/__init__.py
@@ -368,11 +368,9 @@ class ClaudeCodeSolver(CodeSolverStrategy):
'claude',
'-p',
'--output-format',
- 'stream-json',
- #'--max-turns', '100',
- '--debug',
- '--verbose',
- '--dangerously-skip-permissions',
+ 'json',
+ '--max-turns',
+ '10',
]
if CODE_MODEL:
diff --git a/aider_gitea/gitea_client.py b/aider_gitea/gitea_client.py
index bf90768..6c125ec 100644
--- a/aider_gitea/gitea_client.py
+++ b/aider_gitea/gitea_client.py
@@ -15,10 +15,12 @@ class GiteaClient:
Read more about the Gitea API here: https://gitea.com/api/swagger
Attributes:
- gitea_url (str): The base URL for the Gitea API endpoints.
+ ROOT_URL (str): The base URL for the Gitea API endpoints.
session (requests.Session): HTTP session for making API requests.
"""
+ ROOT_URL = None
+
def __init__(self, gitea_url: str, token: str) -> None:
"""Initialize a new Gitea API client.
@@ -30,7 +32,7 @@ class GiteaClient:
AssertionError: If gitea_url ends with '/api/v1'.
"""
assert not gitea_url.endswith('/api/v1')
- self.gitea_url = gitea_url + '/api/v1'
+ self.ROOT_URL = gitea_url + '/api/v1'
self.session = requests.Session()
self.session.headers['Content-Type'] = 'application/json'
if token:
@@ -50,7 +52,7 @@ class GiteaClient:
Raises:
requests.HTTPError: If the API request fails.
"""
- url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/branches/{branch}'
response = self.session.get(url)
response.raise_for_status()
data = response.json()
@@ -71,7 +73,7 @@ class GiteaClient:
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'
+ url = f'{self.ROOT_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:
@@ -93,7 +95,7 @@ class GiteaClient:
Raises:
requests.HTTPError: If the API request fails.
"""
- url = f'{self.gitea_url}/repos/{owner}/{repo}/issues'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/issues'
response = self.session.get(url)
response.raise_for_status()
issues = response.json()
@@ -119,7 +121,7 @@ class GiteaClient:
Returns:
Iterator[str]: An iterator of repository names.
"""
- url = f'{self.gitea_url}/user/repos'
+ url = f'{self.ROOT_URL}/user/repos'
response = self.session.get(url)
response.raise_for_status()
@@ -157,7 +159,7 @@ class GiteaClient:
Raises:
requests.HTTPError: If the API request fails.
"""
- url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls'
json_data = {
'title': title,
'body': body,
@@ -187,7 +189,7 @@ class GiteaClient:
def get_failed_pipelines(self, owner: str, repo: str, pr_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'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs'
response = self.session.get(url)
response.raise_for_status()
runs = response.json().get('workflow_runs', [])
@@ -203,7 +205,7 @@ class GiteaClient:
def get_pipeline_log(self, owner: str, repo: str, run_id: int) -> str:
"""Download the logs for a pipeline run."""
- url = f'{self.gitea_url}/repos/{owner}/{repo}/actions/runs/{run_id}/logs'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/actions/runs/{run_id}/logs'
response = self.session.get(url)
response.raise_for_status()
return response.text
@@ -215,7 +217,112 @@ class GiteaClient:
state: str = 'open',
) -> list[dict]:
"""Fetch pull requests for a repository."""
- url = f'{self.gitea_url}/repos/{owner}/{repo}/pulls?state={state}'
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls?state={state}'
response = self.session.get(url)
response.raise_for_status()
return response.json()
+
+ def get_pull_request_reviews(
+ self,
+ owner: str,
+ repo: str,
+ pr_number: int,
+ ) -> list[dict]:
+ """Download all reviews of a pull request.
+
+ Args:
+ owner (str): Owner of the repository.
+ repo (str): Name of the repository.
+ pr_number (int): Pull request number.
+
+ Returns:
+ list[dict]: A list of review dictionaries.
+
+ Raises:
+ requests.HTTPError: If the API request fails.
+ """
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls/{pr_number}/reviews'
+ response = self.session.get(url)
+ response.raise_for_status()
+ return response.json()
+
+ def get_review_comments(
+ self,
+ owner: str,
+ repo: str,
+ pr_number: int,
+ review_id: int,
+ ) -> list[dict]:
+ """Download all comments for a specific review.
+
+ Args:
+ owner (str): Owner of the repository.
+ repo (str): Name of the repository.
+ pr_number (int): Pull request number.
+ review_id (int): Review ID.
+
+ Returns:
+ list[dict]: A list of review comment dictionaries.
+
+ Raises:
+ requests.HTTPError: If the API request fails.
+ """
+ url = f'{self.ROOT_URL}/repos/{owner}/{repo}/pulls/{pr_number}/reviews/{review_id}/comments'
+ response = self.session.get(url)
+ response.raise_for_status()
+ return response.json()
+
+ def get_pull_request_comments(
+ self,
+ owner: str,
+ repo: str,
+ pr_number: int,
+ ) -> list[dict]:
+ """Download all review comments for a pull request by iterating through all reviews.
+
+ This method implements the proper flow:
+ 1. Download all reviews of the pull request
+ 2. Download all comments for each review
+ 3. Return all comments combined
+
+ Args:
+ owner (str): Owner of the repository.
+ repo (str): Name of the repository.
+ pr_number (int): Pull request number.
+
+ Returns:
+ list[dict]: A list of all review comment dictionaries.
+
+ Raises:
+ requests.HTTPError: If the API request fails.
+ """
+ all_comments = []
+
+ # Step 1: Download all reviews of the pull request
+ reviews = self.get_pull_request_reviews(owner, repo, pr_number)
+
+ # Step 2: Download all comments for each review
+ for review in reviews:
+ review_id = review.get('id')
+ if review_id:
+ try:
+ comments = self.get_review_comments(
+ owner,
+ repo,
+ pr_number,
+ review_id,
+ )
+ # Add review context to each comment
+ for comment in comments:
+ comment['review_id'] = review_id
+ comment['review_state'] = review.get('state')
+ all_comments.extend(comments)
+ except Exception as e:
+ logger.warning(
+ 'Failed to get comments for review %s: %s',
+ review_id,
+ e,
+ )
+ continue
+
+ return all_comments