refactor: extract focused clients from GiteaClient (Slices 1-6)
- Create gitea/issues_client.py with IssuesClient class (9 methods) - Create gitea/prs_client.py with PullRequestsClient class (17 methods) - Create gitea/files_client.py with FilesClient class (4 methods) - Create gitea/notifications_client.py with NotificationsClient class (2 methods) - Create gitea/repos_client.py with ReposClient class (2 methods) - Create gitea/__init__.py to export all client classes - Remove delegation methods from GiteaClient (now ~70 lines) - Update all callers to use sub-clients (client.issues, client.prs, etc.) - Update test files to mock sub-client attributes GiteaClient is now a facade that provides access to focused sub-clients: - repos: Repository operations (ReposClient) - issues: Issue operations (IssuesClient) - prs: Pull request operations (PullRequestsClient) - files: File and git ref operations (FilesClient) - notifications: Notification operations (NotificationsClient) Refs: #godclass-refactor
This commit is contained in:
@@ -0,0 +1,72 @@
|
||||
"""Repositories client for Gitea API operations."""
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
|
||||
from .models import RepositoryModel, UserModel
|
||||
|
||||
|
||||
logger: logging.Logger = logging.getLogger("gitea.repos_client")
|
||||
|
||||
|
||||
class ReposClient:
|
||||
"""HTTP client for Gitea Repositories API operations."""
|
||||
|
||||
def __init__(self, base_url: str, client: httpx.Client, org_filter: str) -> None:
|
||||
"""Initialize the ReposClient.
|
||||
|
||||
Args:
|
||||
base_url: The base URL for the Gitea API.
|
||||
client: The httpx client for making requests.
|
||||
org_filter: Organization filter for repositories.
|
||||
"""
|
||||
self.base_url: str = base_url
|
||||
self.client: httpx.Client = client
|
||||
self.org_filter: str = org_filter
|
||||
|
||||
def list_all_user_repos(self) -> list[RepositoryModel]:
|
||||
"""List all repositories for the authenticated user.
|
||||
|
||||
Returns:
|
||||
List of repositories belonging to the configured organization.
|
||||
"""
|
||||
try:
|
||||
url = f"{self.base_url}/api/v1/user/repos"
|
||||
response = self.client.get(url)
|
||||
response.raise_for_status()
|
||||
repos: list[dict[str, Any]] = response.json()
|
||||
# Filter to ONLY configured organization repos, include mirrors
|
||||
seen: set[str] = set()
|
||||
result: list[RepositoryModel] = []
|
||||
for r in repos:
|
||||
full_name = r.get("full_name", "")
|
||||
if (
|
||||
full_name
|
||||
and full_name not in seen
|
||||
and (r.get("owner") or {}).get("login") == self.org_filter
|
||||
):
|
||||
seen.add(full_name)
|
||||
result.append(RepositoryModel(**r))
|
||||
return result
|
||||
except Exception as e:
|
||||
logger.error(f"Error listing user repos: {e}", exc_info=True)
|
||||
return []
|
||||
|
||||
def get_authenticated_user(self) -> UserModel:
|
||||
"""Get the authenticated user.
|
||||
|
||||
Returns:
|
||||
The authenticated user.
|
||||
|
||||
Raises:
|
||||
RuntimeError: If the user cannot be retrieved.
|
||||
"""
|
||||
try:
|
||||
response = self.client.get(f"{self.base_url}/api/v1/user")
|
||||
response.raise_for_status()
|
||||
return UserModel(**response.json())
|
||||
except Exception as e:
|
||||
logger.error(f"Error getting authenticated user: {e}", exc_info=True)
|
||||
raise RuntimeError(f"Could not get authenticated user: {e}") from e
|
||||
Reference in New Issue
Block a user