Remove more unused code

services/__init__.py

@@ -182,23 +182,23 @@ def _not_found(self, url: str, tag: str) -> Issue:
             raw={},
         )
 
-    def _get_user_issues_x(self, queries: list[Any], **kwargs) -> list[Issue]:
+    def _get_user_issues_x(self, queries: list[Any]) -> list[Issue]:
         issues = []
         futures = []
         with concurrent.futures.ThreadPoolExecutor(
             max_workers=len(queries)
         ) as executor:
             for query in queries:
-                futures.append(executor.submit(self._get_user_issues, query, **kwargs))
+                futures.append(executor.submit(self._get_user_issues, query))
         for future in concurrent.futures.as_completed(futures):
             issues.extend(future.result())
         return list(set(issues))
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         raise NotImplementedError("_get_user_issues()")
 
     @abstractmethod
-    def get_user_issues(self, **_) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         """
         Get user issues
         """

services/bugzilla.py

@@ -40,28 +40,25 @@ def close(self):
         except (AttributeError, BugzillaError):
             pass
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         try:
             issues = self.client.query(query)
         except (AttributeError, BugzillaError, RequestException) as exc:
             logging.error("Bugzilla: %s: get_user_issues(): %s", self.url, exc)
             return []
         return [self._to_issue(issue) for issue in issues if issue.is_open]
 
-    def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         try:
-            username = username or self.client.user
-            user = self.client.getuser(username)
+            user = self.client.getuser(self.client.user)
         except (AttributeError, BugzillaError, RequestException) as exc:
-            logging.error(
-                "Bugzilla: %s: get_user_issues(%s): %s", self.url, username, exc
-            )
+            logging.error("Bugzilla: %s: get_user_issues(): %s", self.url, exc)
             return []
         queries = [
             {"assigned_to": user.email},
             {"reporter": user.email},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         try:

services/gitea.py

@@ -32,12 +32,8 @@ def _get_paginated(self, url: str, params: dict[str, str] | None) -> list[dict]:
             params = {}
         if "limit" not in params:
             params["limit"] = "100"
-        try:
-            got = self.session.get(url, params=params)
-            got.raise_for_status()
-        except RequestException as exc:
-            logging.error("Gitea: %s: Error while fetching page 1: %s", url, exc)
-            raise
+        got = self.session.get(url, params=params)
+        got.raise_for_status()
         entries: list[dict] = got.json()
         if "Link" in got.headers:
             links = parse_header_links(got.headers["Link"])
@@ -78,27 +74,30 @@ def get_page(page: int) -> list[dict]:
 
         return entries
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         query["state"] = "open"
-        issues = self._get_paginated(
-            f"{self.url}/api/v1/repos/issues/search", params=query
-        )
+        try:
+            issues = self._get_paginated(
+                f"{self.url}/api/v1/repos/issues/search", params=query
+            )
+        except RequestException as exc:
+            logging.error("Gitea: %s: get_user_issues(): %s", self.url, exc)
+            return []
         return [
             self._to_issue(issue, is_pr=bool(issue["pull_request"])) for issue in issues
         ]
 
-    def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         # Not possible to filter issues by username because of:
         # https://github.com/go-gitea/gitea/issues/25979
-        _ = username
         queries = [
             {"assigned": True},
             {"created": True},
             {"mentioned": True},
             {"reviewed": True},
             {"review_requested": True},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def _to_issue(self, info: Any, **kwargs) -> Issue:
         repo = kwargs.get("repo", info["repository"]["full_name"])

services/github.py

@@ -36,15 +36,13 @@ def close(self):
         except (AttributeError, GithubException):
             pass
 
-    def get_user_issues(self, username: str = "", **_) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         try:
-            user = (
-                self.client.get_user(username) if username else self.client.get_user()
-            )
+            user = self.client.get_user()
             filters = f"state:open involves:{user.login}"
             issues = self.client.search_issues(filters)
         except (GithubException, RequestException) as exc:
-            logging.error("Github: get_user_issues(%s): %s", username, exc)
+            logging.error("Github: get_user_issues(): %s", exc)
             return []
 
         return [

services/gitlab.py

@@ -45,7 +45,7 @@ def close(self):
         except (AttributeError, GitlabError):
             pass
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         issues: list[Any] = []
         query |= {
             "all": True,  # No pagination
@@ -62,18 +62,13 @@ def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
             return []
         return [self._to_issue(issue) for issue in issues]
 
-    def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         try:
-            if username:
-                user = self.client.users.list(username=username)[0]  # type: ignore
-            else:
-                user = self.client.user
+            user = self.client.user
             if user is None:
                 return []
         except (GitlabError, RequestException) as exc:
-            logging.error(
-                "Gitlab: %s: get_user_issues(%s): %s", self.url, username, exc
-            )
+            logging.error("Gitlab: %s: get_user_issues(): %s", self.url, exc)
             return []
         queries = [
             {"assignee_id": user.id, "pull_requests": False},
@@ -82,7 +77,7 @@ def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
             {"author_id": user.id, "pull_requests": True},
             {"reviewer_id": user.id, "pull_requests": True},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         repo: str = kwargs.pop("repo")

services/jira.py

@@ -42,13 +42,12 @@ def _get_issues(self, filters: str) -> list[dict]:
             issues.extend(data["issues"])
         return issues
 
-    def get_user_issues(self, username: str = "", **_) -> list[Issue]:
-        username = username or self.client.username or "currentUser()"
-        filters = f"watcher = {username}"
+    def get_user_issues(self) -> list[Issue]:
+        filters = "watcher = currentUser()"
         try:
             issues = self._get_issues(filters)
         except (ApiError, RequestException) as exc:
-            logging.error("Jira: %s: get_user_issues(%s): %s", self.url, username, exc)
+            logging.error("Jira: %s: get_user_issues(): %s", self.url, exc)
             return []
         return [self._to_issue(issue) for issue in issues]
 

services/launchpad.py

@@ -32,7 +32,7 @@ def __init__(self, url: str, creds: dict):
         self.tag = "lp"
         self.username = creds.get("username", "")
 
-    def _get_user_issues(self, query: dict[str, Any], **_) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         query["ws.op"] = "searchTasks"
         try:
             got = self.session.get("https://api.launchpad.net/1.0/bugs", params=query)
@@ -43,14 +43,13 @@ def _get_user_issues(self, query: dict[str, Any], **_) -> list[Issue]:
             return []
         return [self._to_issue(issue) for issue in issues]
 
-    def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
-        username = username or self.username
-        user_url = f"https://api.launchpad.net/1.0/~{username}"
+    def get_user_issues(self) -> list[Issue]:
+        user_url = f"https://api.launchpad.net/1.0/~{self.username}"
         queries = [
             {"assignee": user_url},
             {"bug_reporter": user_url},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         if not kwargs.get("repo"):

services/pagure.py

@@ -90,50 +90,43 @@ def get_page(page: int) -> list[dict]:
 
         return entries
 
-    def _get_issues(self, username: str, **params) -> list[dict]:
+    def _get_issues(self, **params) -> list[dict]:
         if params["assignee"]:
             key = "issues_assigned"
             next_key = "pagination_issues_assigned"
         else:
             key = "issues_created"
             next_key = "pagination_issues_created"
-        url = f"{self.url}/api/0/user/{username}/issues"
+        url = f"{self.url}/api/0/user/{self.username}/issues"
         return self._get_paginated(url, params=params, key=key, next_key=next_key)
 
-    def _get_pullrequests(
-        self, username: str, created: bool = False, **params
-    ) -> list[dict]:
+    def _get_pullrequests(self, created: bool = False, **params) -> list[dict]:
         pr_type = "filed" if created else "actionable"
-        url = f"{self.url}/api/0/user/{username}/requests/{pr_type}"
+        url = f"{self.url}/api/0/user/{self.username}/requests/{pr_type}"
         return self._get_paginated(
             url, params=params, key="requests", next_key="pagination"
         )
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         pull_requests = query.pop("pull_requests")
-        username = kwargs.pop("username")
         try:
             if pull_requests:
-                issues = self._get_pullrequests(username, **query)
+                issues = self._get_pullrequests(**query)
             else:
-                issues = self._get_issues(username, **query)
+                issues = self._get_issues(**query)
         except RequestException as exc:
             logging.error("Pagure: %s: get_user_issues(): %s", self.url, exc)
             return []
         return [self._to_issue(issue, is_pr=pull_requests) for issue in issues]
 
-    def get_user_issues(self, username: str = "", **kwargs) -> list[Issue]:
-        username = username or self.username
-        if not username:
-            return []
-        kwargs["username"] = username
+    def get_user_issues(self) -> list[Issue]:
         queries = [
             {"pull_requests": False, "assignee": 0, "author": 1},
             {"pull_requests": False, "assignee": 1, "author": 0},
             {"pull_requests": True, "created": True},
             {"pull_requests": True, "created": False},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def _to_issue(self, info: Any, **kwargs) -> Issue:
         repo = kwargs.get("repo", "") or info["project"]["fullname"]

services/redmine.py

@@ -36,27 +36,25 @@ def close(self):
         except AttributeError:
             pass
 
-    def _get_user_issues(self, query: dict[str, Any], **kwargs) -> list[Issue]:
+    def _get_user_issues(self, query: dict[str, Any]) -> list[Issue]:
         try:
             issues = self.client.issue.filter(**query)
         except (BaseRedmineError, RequestException) as exc:
             logging.error("Redmine: %s: get_user_issues(): %s", self.url, exc)
             return []
         return [self._to_issue(issue) for issue in issues]
 
-    def get_user_issues(self, username: str = "me", **kwargs) -> list[Issue]:
+    def get_user_issues(self) -> list[Issue]:
         try:
-            user = self.client.user.get(username)
+            user = self.client.user.get("me")
         except (BaseRedmineError, RequestException) as exc:
-            logging.error(
-                "Redmine: %s: get_user_issues(%s): %s", self.url, username, exc
-            )
+            logging.error("Redmine: %s: get_user_issues(): %s", self.url, exc)
             return []
         queries = [
             {"assigned_to_id": user.id},
             {"author_id": user.id},
         ]
-        return self._get_user_issues_x(queries, **kwargs)
+        return self._get_user_issues_x(queries)
 
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         try: