feat: implement Gitea notification polling and state persistence

Squash merged Gitea notification polling and state persistence.
This commit was merged in pull request #1.
This commit is contained in:
2026-06-30 21:31:59 +02:00
parent edc8415571
commit 287d20bc56
4 changed files with 281 additions and 39 deletions
+34 -1
View File
@@ -1,7 +1,7 @@
import httpx
import json
import base64
from typing import Any
from typing import Any, Optional
from .config import GITEA_URL, GITEA_TOKEN
from .models import (
UserModel,
@@ -396,3 +396,36 @@ class GiteaClient(IssuesClient, PullRequestsClient, FilesClient, RefsClient, Rep
if isinstance(data, list):
return [item.get("content", "") for item in data if item.get("type") == "file"]
return base64.b64decode(data.get("content", "")).decode() if data.get("content") else ""
def list_unread_notifications(self, since: Optional[str] = None) -> list[dict[str, Any]]:
try:
with httpx.Client() as client:
url = f"{self.base_url}/api/v1/notifications?all=false"
if since:
url += f"&since={since}"
response = client.get(url, headers=self.headers)
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 == "meeks":
result.append(n)
return result
except Exception as e:
print(f"Error listing unread notifications: {e}")
return []
def mark_notification_as_read(self, thread_id: int) -> bool:
try:
with httpx.Client() as client:
url = f"{self.base_url}/api/v1/notifications/threads/{thread_id}"
response = client.patch(url, headers=self.headers)
response.raise_for_status()
return True
except Exception as e:
print(f"Error marking notification thread {thread_id} as read: {e}")
return False