meeks
ae4e2d46ac
- Replace Any with object or specific types across codebase - Add ReviewRequest dataclass for PR review payloads - Update bad_code.md: mark 5.1 (Any Type Overuse) as resolved - Fix summary table with accurate counts and unresolved issues list
72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
"""Repositories client for Gitea API operations."""
|
|
|
|
import logging
|
|
|
|
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, object]] = 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
|