133 lines
4.5 KiB
Python
133 lines
4.5 KiB
Python
import logging
|
|
from collections.abc import Iterator
|
|
|
|
import requests
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GiteaClient:
|
|
"""
|
|
Client for interacting with the Gitea API.
|
|
|
|
This class provides methods to interact with a Gitea instance's API,
|
|
including retrieving repository information, creating branches, and fetching issues.
|
|
|
|
Attributes:
|
|
gitea_url (str): The base URL for the Gitea API endpoints.
|
|
session (requests.Session): HTTP session for making API requests.
|
|
"""
|
|
|
|
def __init__(self, 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.
|
|
|
|
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'
|
|
if token:
|
|
self.session.headers['Authorization'] = f'token {token}'
|
|
|
|
def get_default_branch_sha(self, owner: str, repo: str, branch: 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.
|
|
|
|
Returns:
|
|
str: 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)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
return data['commit']['sha']
|
|
|
|
def create_branch(self, owner: str, repo: str, new_branch: str, 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.
|
|
|
|
Returns:
|
|
bool: 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(f'Branch {new_branch} already exists.')
|
|
return False
|
|
response.raise_for_status()
|
|
return True
|
|
|
|
def get_issues(self, owner: str, repo: str) -> list[dict[str, str]]:
|
|
"""
|
|
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.
|
|
|
|
Returns:
|
|
list: A list of issue dictionaries, 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)
|
|
response.raise_for_status()
|
|
issues = 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', []))
|
|
]
|
|
return issues
|
|
|
|
def iter_user_repositories(
|
|
self, owner: 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.
|
|
|
|
Returns:
|
|
list: A list of repository names.
|
|
"""
|
|
url = f'{self.gitea_url}/user/repos'
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
|
|
for repo in response.json():
|
|
if only_those_with_issues and not repo['has_issues']:
|
|
continue
|
|
if repo['owner']['login'].lower() != owner.lower():
|
|
continue
|
|
yield repo['name']
|