meeks
25473ed684
- 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
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Notifications client for Gitea API operations."""
|
|
|
|
import logging
|
|
from typing import Any, Optional
|
|
|
|
import httpx
|
|
|
|
|
|
logger: logging.Logger = logging.getLogger("gitea.notifications_client")
|
|
|
|
|
|
class NotificationsClient:
|
|
"""HTTP client for Gitea Notifications API operations."""
|
|
|
|
def __init__(self, base_url: str, client: httpx.Client, org_filter: str) -> None:
|
|
"""Initialize the NotificationsClient.
|
|
|
|
Args:
|
|
base_url: The base URL for the Gitea API.
|
|
client: The httpx client for making requests.
|
|
org_filter: Organization filter for notifications.
|
|
"""
|
|
self.base_url: str = base_url
|
|
self.client: httpx.Client = client
|
|
self.org_filter: str = org_filter
|
|
|
|
def list_unread_notifications(
|
|
self, since: Optional[str] = None
|
|
) -> list[dict[str, Any]]:
|
|
"""List unread notifications.
|
|
|
|
Args:
|
|
since: Optional ISO 8601 timestamp to filter notifications after.
|
|
|
|
Returns:
|
|
List of unread notifications for the configured organization.
|
|
"""
|
|
try:
|
|
url = f"{self.base_url}/api/v1/notifications"
|
|
params: dict[str, str] = {"all": "false"}
|
|
if since:
|
|
params["since"] = since
|
|
response = self.client.get(url, params=params)
|
|
response.raise_for_status()
|
|
notifications: list[dict[str, Any]] = response.json()
|
|
|
|
result: list[dict[str, Any]] = []
|
|
for n in notifications:
|
|
repo_info = n.get("repository") or {}
|
|
owner_info = repo_info.get("owner") or {}
|
|
owner_login = owner_info.get("login", "")
|
|
if owner_login == self.org_filter:
|
|
result.append(n)
|
|
return result
|
|
except Exception as e:
|
|
logger.error(f"Error listing unread notifications: {e}", exc_info=True)
|
|
return []
|
|
|
|
def mark_notification_as_read(self, thread_id: int) -> bool:
|
|
"""Mark a notification as read.
|
|
|
|
Args:
|
|
thread_id: Notification thread ID.
|
|
|
|
Returns:
|
|
True if successful, False otherwise.
|
|
"""
|
|
try:
|
|
url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}"
|
|
response = self.client.patch(url)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception as e:
|
|
logger.error(
|
|
f"Error marking notification thread {thread_id} as read: {e}",
|
|
exc_info=True,
|
|
)
|
|
return False
|