implement support for Forgejo comments and reactions

Author: julietowah

ogr/abstract.py

@@ -17,13 +17,11 @@
 
 import github
 import gitlab
-import pyforgejo
 import requests
 
 from ogr.deprecation import deprecate_and_set_removal
 from ogr.exceptions import (
     APIException,
-    ForgejoAPIException,
     GitForgeInternalError,
     GithubAPIException,
     GitlabAPIException,
@@ -61,11 +59,7 @@ def __check_for_internal_failure(ex: APIException):
 
 
 def __wrap_exception(
-    ex: Union[
-        github.GithubException,
-        gitlab.GitlabError,
-        pyforgejo.core.api_error.ApiError,
-    ],
+    ex: Union[github.GithubException, gitlab.GitlabError],
 ) -> APIException:
     """
     Wraps uncaught exception in one of ogr exceptions.
@@ -82,7 +76,6 @@ def __wrap_exception(
     MAPPING = {
         github.GithubException: GithubAPIException,
         gitlab.GitlabError: GitlabAPIException,
-        pyforgejo.core.api_error.ApiError: ForgejoAPIException,
     }
 
     for caught_exception, ogr_exception in MAPPING.items():
@@ -121,11 +114,7 @@ def wrapper(*args, **kwargs):
             ) from ex
         except APIException as ex:
             __check_for_internal_failure(ex)
-        except (
-            github.GithubException,
-            gitlab.GitlabError,
-            pyforgejo.core.api_error.ApiError,
-        ) as ex:
+        except (github.GithubException, gitlab.GitlabError) as ex:
             __check_for_internal_failure(__wrap_exception(ex))
 
     return wrapper
@@ -168,34 +157,15 @@ def __repr__(self) -> str:
 
 
 class Reaction(OgrAbstractClass):
-    def __init__(
-        self,
-        raw_reaction: Any,
-        reaction_id: Optional[int] = None,
-        parent: Optional[Any] = None,
-    ) -> None:
+    def __init__(self, raw_reaction: Any) -> None:
         self._raw_reaction = raw_reaction
-        self._id = reaction_id
-        self._parent = parent
 
     def __str__(self):
         return f"Reaction(raw_reaction={self._raw_reaction})"
 
     def delete(self) -> None:
         """Delete a reaction."""
-        if not self._id:
-            raise ValueError("Cannot delete a reaction without an ID.")
-        if not self._parent:
-            raise ValueError("Cannot delete a reaction without a parent object.")
-        if not self._raw_reaction or "content" not in self._raw_reaction:
-            raise ValueError("Invalid raw reaction data.")
-
-        self._parent.client.issue.delete_comment_reaction(
-            owner=self._parent.repo_namespace,
-            repo=self._parent.repo_name,
-            id=self._id,
-            content=self._raw_reaction["content"],
-        )
+        raise NotImplementedError()
 
 
 class Comment(OgrAbstractClass):
@@ -234,11 +204,7 @@ def __str__(self) -> str:
 
     def _from_raw_comment(self, raw_comment: Any) -> None:
         """Constructs Comment object from raw_comment given from API."""
-        self._id = raw_comment["id"]
-        self._body = raw_comment["body"]
-        self._author = raw_comment["user"]["login"]
-        self._created = raw_comment["created_at"]
-        self._edited = raw_comment.get("updated_at")
+        raise NotImplementedError()
 
     @property
     def body(self) -> str:
@@ -270,15 +236,7 @@ def edited(self) -> datetime.datetime:
 
     def get_reactions(self) -> list[Reaction]:
         """Returns list of reactions."""
-        if not self._id or not self._parent:
-            raise ValueError("Cannot fetch reactions without a comment ID and parent.")
-
-        reactions_data = self._parent.client.issue.get_comment_reactions(
-            owner=self._parent.repo_namespace,
-            repo=self._parent.repo_name,
-            id=self._id,
-        )
-        return [Reaction(raw_reaction=react) for react in reactions_data]
+        raise NotImplementedError()
 
     def add_reaction(self, reaction: str) -> Reaction:
         """
@@ -292,16 +250,7 @@ def add_reaction(self, reaction: str) -> Reaction:
         Returns:
             Object representing newly added reaction.
         """
-        if not self._id or not self._parent:
-            raise ValueError("Cannot add reaction without a comment ID and parent.")
-
-        reaction_data = self._parent.client.issue.add_comment_reaction(
-            owner=self._parent.repo_namespace,
-            repo=self._parent.repo_name,
-            id=self._id,
-            content=reaction,
-        )
-        return Reaction(raw_reaction=reaction_data)
+        raise NotImplementedError()
 
 
 class IssueComment(Comment):
@@ -324,6 +273,125 @@ def __str__(self) -> str:
         return "PR" + super().__str__()
 
 
+class ForgejoAPI:
+
+    def __init__(self, base_url, token):
+        self.base_url = base_url.rstrip("/")
+        self.headers = {
+            "Authorization": f"token {token}",
+            "Content-Type": "application/json",
+        }
+
+    def create_comment(self, owner, repo, issue_index, body):
+        """Create a comment on an issue/pull request"""
+        url = (
+            f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_index}/comments"
+        )
+        payload = {"body": body}
+        response = requests.post(url, headers=self.headers, json=payload)
+        response.raise_for_status()
+        return response.json()
+
+    def list_comments(self, owner, repo, issue_index):
+        """List all comments on an issue/pull request"""
+        url = (
+            f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/{issue_index}/comments"
+        )
+        response = requests.get(url, headers=self.headers)
+        response.raise_for_status()
+        return response.json()
+
+    def add_reaction(self, owner, repo, comment_id, reaction_type):
+        """Add a reaction to a comment"""
+        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/comments/{comment_id}/reactions"
+        payload = {"content": reaction_type}
+        response = requests.post(url, headers=self.headers, json=payload)
+        response.raise_for_status()
+        return response.json()
+
+    def delete_reaction(self, owner, repo, comment_id, reaction_type):
+        """Remove a reaction from a comment"""
+        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/comments/{comment_id}/reactions"
+        params = {"content": reaction_type}
+        response = requests.delete(url, headers=self.headers, params=params)
+        response.raise_for_status()
+        return response.json()
+
+    def list_reactions(self, owner, repo, comment_id):
+        """List reactions on a comment"""
+        url = f"{self.base_url}/api/v1/repos/{owner}/{repo}/issues/comments/{comment_id}/reactions"
+        response = requests.get(url, headers=self.headers)
+        response.raise_for_status()
+        return response.json()
+
+
+class ForgejoReaction(Reaction):
+    def __init__(self, raw_reaction, api_client, owner, repo, comment_id):
+        super().__init__(raw_reaction)
+        self._api = api_client
+        self._owner = owner
+        self._repo = repo
+        self._comment_id = comment_id
+        self.reaction_type = raw_reaction.get("content")
+
+    def delete(self) -> None:
+        self._api.delete_reaction(
+            self._owner,
+            self._repo,
+            self._comment_id,
+            self.reaction_type,
+        )
+
+
+class ForgejoIssueComment(IssueComment):
+    def _from_raw_comment(self, raw_comment: dict) -> None:
+        self._body = raw_comment.get("body")
+        self._id = raw_comment.get("id")
+        self._author = raw_comment.get("user", {}).get("login")
+        self._created = datetime.datetime.fromisoformat(raw_comment.get("created_at"))
+        self._edited = (
+            datetime.datetime.fromisoformat(raw_comment["updated_at"])
+            if raw_comment.get("updated_at")
+            else None
+        )
+
+    def get_reactions(self) -> list[Reaction]:
+        owner = self._parent.project.owner
+        repo = self._parent.project.repo
+        reactions = self._parent.service.forgejo_api.list_reactions(
+            owner,
+            repo,
+            self.id,
+        )
+        return [
+            ForgejoReaction(
+                raw_reaction=r,
+                api_client=self._parent.service.forgejo_api,
+                owner=owner,
+                repo=repo,
+                comment_id=self.id,
+            )
+            for r in reactions
+        ]
+
+    def add_reaction(self, reaction: str) -> ForgejoReaction:
+        owner = self._parent.project.owner
+        repo = self._parent.project.repo
+        reaction_data = self._parent.service.forgejo_api.add_reaction(
+            owner,
+            repo,
+            self.id,
+            reaction,
+        )
+        return ForgejoReaction(
+            raw_reaction=reaction_data,
+            api_client=self._parent.service.forgejo_api,
+            owner=owner,
+            repo=repo,
+            comment_id=self.id,
+        )
+
+
 class IssueStatus(IntEnum):
     """Enumeration for issue statuses."""
 

recipe.yaml

@@ -27,6 +27,7 @@
         name:
           - twine # we need newest twine, b/c of the check command
           - readme_renderer[md]
+          - pytest-cov
 
         state: latest
       become: true