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
79 lines
2.5 KiB
Python
79 lines
2.5 KiB
Python
"""Notifications client for Gitea API operations."""
|
|
|
|
import logging
|
|
from typing import 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, object]]:
|
|
"""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, object]] = response.json()
|
|
|
|
result: list[dict[str, object]] = []
|
|
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
|