Fix MyGitea

services/gitea.py

@@ -3,7 +3,6 @@
 """
 
 import logging
-from concurrent.futures import ThreadPoolExecutor
 from typing import Any
 
 from requests.utils import parse_header_links
@@ -14,8 +13,6 @@
 
 
 # Reference: https://try.gitea.io/api/swagger
-# Not possible to filter issues because of:
-# https://github.com/go-gitea/gitea/issues/25979
 class MyGitea(Generic):
     """
     Gitea
@@ -47,90 +44,78 @@ def _get_paginated(self, url: str, params: dict[str, str]) -> list[dict]:
             break
         return entries
 
+    # Not possible to filter issues by username because of:
+    # https://github.com/go-gitea/gitea/issues/25979
+
     def _get_issues(  # pylint: disable=too-many-arguments
         self,
-        issue_type: str = "issues",
         assigned: bool = False,
         created: bool = False,
-        involved: bool = False,
         closed: bool = False,
+        pull_requests: bool = False,
     ) -> list[Issue]:
         params: dict[str, Any] = {
             "state": "closed" if closed else "open",
-            "type": issue_type,
+            "type": "pulls" if pull_requests else "issues",
         }
-        # Missing: review_requested & reviewed
+        # Missing: mentioned, review_requested & reviewed
         if assigned:
             params["assigned"] = True
         if created:
             params["created"] = True
-        if involved:
-            params["mentioned"] = True
         issues = self._get_paginated(
             f"{self.url}/api/v1/repos/issues/search", params=params
         )
-        return [
-            self._to_issue(issue, is_pr=bool(issue_type == "pulls")) for issue in issues
-        ]
+        return [self._to_issue(issue, is_pr=pull_requests) for issue in issues]
 
     def get_assigned(
-        self, username: str = "", state: str = "open", **_
+        self, username: str = "", pull_requests: bool = False, state: str = "open", **_
     ) -> list[Issue] | None:
         """
         Get assigned issues
         """
-        return self.get_user_issues(
-            username, assigned=True, involved=False, state=state
-        )
+        username = ""
+        try:
+            return self._get_issues(
+                assigned=True, closed=bool(state != "open"), pull_requests=pull_requests
+            )
+        except RequestException as exc:
+            logging.error("Gitea: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
 
     def get_created(
-        self, username: str = "", state: str = "open", **_
+        self, username: str = "", pull_requests: bool = False, state: str = "open", **_
     ) -> list[Issue] | None:
         """
         Get created issues
         """
-        return self.get_user_issues(username, created=True, involved=False, state=state)
+        username = ""
+        try:
+            return self._get_issues(
+                created=True, closed=bool(state != "open"), pull_requests=pull_requests
+            )
+        except RequestException as exc:
+            logging.error("Gitea: %s: get_created(%s): %s", self.url, username, exc)
+        return None
 
     def get_user_issues(  # pylint: disable=too-many-arguments
         self,
         username: str = "",
         assigned: bool = False,
         created: bool = False,
         involved: bool = True,
-        closed: bool = False,
-        **_,
+        **kwargs,
     ) -> list[Issue] | None:
         """
         Get user issues
         """
-        kwargs = {"closed": closed}
-        if assigned:
-            kwargs["assigned"] = True
-        elif created:
-            kwargs["created"] = True
-        elif involved:
-            kwargs["assigned"] = True
-            kwargs["created"] = True
-            # kwargs["involved"] = True
-        all_issues: list[Issue] = []
-
-        def get_issues(issue_type: str) -> list[Issue] | None:
-            try:
-                return self._get_issues(issue_type=issue_type, **kwargs)
-            except RequestException as exc:
-                logging.error(
-                    "Gitea: %s: get_user_issues(%s): %s", self.url, username, exc
-                )
-            return None
-
-        with ThreadPoolExecutor(max_workers=2) as executor:
-            results = executor.map(get_issues, ("issues", "pulls"))
-            for result in results:
-                if result is None:
-                    return None
-                all_issues.extend(result)
-
-        return all_issues
+        return self._get_user_issues4(
+            username=username,
+            assigned=assigned,
+            created=created,
+            involved=involved,
+            **kwargs,
+        )
 
     def _to_issue(self, info: Any, **kwargs) -> Issue:
         repo = kwargs.get("repo", info["repository"]["full_name"])