Introduce get_created() & get_assigned()

creds-example.json

@@ -1,12 +1,15 @@
 {
   "bugzilla.suse.com": {
+    "user": "SUSE_BUGZILLA_USER",
     "api_key": "SUSE_BUGZILLA_API_KEY"
   },
   "bugzilla.redhat.com": {
     "api_key": "REDHAT_BUGZILLA_API_KEY"
   },
   "jira.suse.com": {
+    "username": "JIRA_USER",
     "cookies": {
+      "seraph.rememberme.cookie": "JIRA_COOKIE",
       "JIRASESSIONID": "JIRA_SESSION_ID",
       "atlassian.xsrf.token": "JIRA_TOKEN"
     }
@@ -28,5 +31,8 @@
   },
   "code.opensuse.org": {
     "token": "PAGURE_TOKEN"
+  },
+  "launchpad.net": {
+    "username": "LAUNCHPAD_USER"
   }
 }

services/bugzilla.py

@@ -14,6 +14,7 @@
 from . import Service, Issue, debugme, status
 
 
+# Reference: https://bugzilla.readthedocs.io/en/latest/api/index.html#apis
 class MyBugzilla(Service):
     """
     Bugzilla
@@ -39,6 +40,34 @@ def __del__(self):
         except (AttributeError, BugzillaError):
             pass
 
+    def _get_issues(self, filters: dict) -> list[Issue]:
+        return [
+            self._to_issue(bug) for bug in self.client.query(filters) if bug.is_open
+        ]
+
+    def get_assigned(self, username: str = "") -> list[Issue] | None:
+        """
+        Get assigned Bugzilla issues
+        """
+        username = username or self.client.user
+        try:
+            user = self.client.getuser(username)
+            return self._get_issues({"assigned_to": user.email})
+        except (AttributeError, BugzillaError, RequestException) as exc:
+            logging.error("Bugzilla: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
+
+    def get_created(self, username: str = "") -> list[Issue] | None:
+        """
+        Get created Bugzilla issues
+        """
+        try:
+            user = self.client.getuser(username)
+            return self._get_issues({"creator": user.email})
+        except (AttributeError, BugzillaError, RequestException) as exc:
+            logging.error("Bugzilla: %s: get_created(%s): %s", self.url, username, exc)
+        return None
+
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         """
         Get Bugzilla issue

services/gitea.py

@@ -9,6 +9,8 @@
 
 
 # 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
@@ -23,7 +25,7 @@ def __init__(self, url: str, creds: dict):
 
     def _to_issue(self, info: Any, **kwargs) -> Issue:
         repo = kwargs.pop("repo")
-        is_pr = kwargs.pop("is_pr")
+        is_pr = bool(kwargs.get("is_pr"))
         mark = "!" if is_pr else "#"
         return Issue(
             tag=f'{self.tag}#{repo}{mark}{info["number"]}',

services/github.py

@@ -13,6 +13,8 @@
 from . import Service, Issue, status
 
 
+# Reference:
+# https://docs.github.com/en/search-github/searching-on-github/searching-issues-and-pull-requests
 class MyGithub(Service):
     """
     Github
@@ -34,6 +36,44 @@ def __del__(self):
         except (AttributeError, GithubException):
             pass
 
+    def _get_issues(self, filters: str, is_pr: bool) -> list[Issue]:
+        issue_type = "pr" if is_pr else "issue"
+        filters = f"{filters} type:{issue_type}"
+        return [
+            self._to_issue(issue, is_pr=is_pr)
+            for issue in self.client.search_issues(filters)
+        ]
+
+    def get_assigned(
+        self, username: str | None = None, is_pr: bool = False, state: str = "open"
+    ) -> list[Issue] | None:
+        """
+        Get assigned Github issues
+        """
+        try:
+            user = (
+                self.client.get_user(username) if username else self.client.get_user()
+            )
+            return self._get_issues(f"assignee:{user.login} state:{state}", is_pr)
+        except (GithubException, RequestException) as exc:
+            logging.error("Github: get_assigned(%s): %s", username, exc)
+        return None
+
+    def get_created(
+        self, username: str | None = None, is_pr: bool = False, state: str = "open"
+    ) -> list[Issue] | None:
+        """
+        Get created Github issues
+        """
+        try:
+            user = (
+                self.client.get_user(username) if username else self.client.get_user()
+            )
+            return self._get_issues(f"author:{user.login} state:{state}", is_pr)
+        except (GithubException, RequestException) as exc:
+            logging.error("Github: get_created(%s): %s", username, exc)
+        return None
+
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         """
         Get Github issue
@@ -59,7 +99,8 @@ def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
             return None
         return self._to_issue(info, repo, is_pr)
 
-    def _to_issue(self, info: Any, repo: str, is_pr: bool) -> Issue:
+    def _to_issue(self, info: Any, repo: str = "", is_pr: bool = False) -> Issue:
+        repo = repo or info.repository.full_name
         mark = "!" if is_pr else "#"
         return Issue(
             tag=f"{self.tag}#{repo}{mark}{info.number}",

services/gitlab.py

@@ -15,6 +15,9 @@
 from . import Service, Issue, debugme, status
 
 
+# References:
+# https://docs.gitlab.com/ee/api/issues.html
+# https://docs.gitlab.com/ee/api/merge_requests.html
 class MyGitlab(Service):
     """
     Gitlab
@@ -42,6 +45,54 @@ def __del__(self):
         except (AttributeError, GitlabError):
             pass
 
+    def _get_issues(self, **filters) -> list[Issue]:
+        return [
+            self._to_issue(issue)
+            for issue in self.client.issues.list(all=True, **filters)
+        ]
+
+    def _get_pullrequests(self, **filters) -> list[Issue]:
+        return [
+            self._to_issue(issue)
+            for issue in self.client.mergerequests.list(all=True, **filters)
+        ]
+
+    def get_assigned(
+        self, username: str | None = None, is_pr: bool = False, state: str = "opened"
+    ) -> list[Issue] | None:
+        """
+        Get assigned Github issues
+        """
+        try:
+            if username:
+                user = self.client.users.list(username=username)[0]  # type: ignore
+            else:
+                user = self.client.user
+            if is_pr:
+                return self._get_pullrequests(assignee_id=user.id, state=state)
+            return self._get_issues(assignee_id=user.id, state=state)
+        except (GitlabError, RequestException) as exc:
+            logging.error("Gitlab: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
+
+    def get_created(
+        self, username: str | None = None, is_pr: bool = False, state: str = "opened"
+    ) -> list[Issue] | None:
+        """
+        Get created Github issues
+        """
+        try:
+            if username:
+                user = self.client.users.list(username=username)[0]  # type: ignore
+            else:
+                user = self.client.user
+            if is_pr:
+                return self._get_pullrequests(author=user.id, state=state)
+            return self._get_issues(author=user.id, state=state)
+        except (GitlabError, RequestException) as exc:
+            logging.error("Gitlab: %s: get_created(%s): %s", self.url, username, exc)
+        return None
+
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         """
         Get Gitlab issue

services/jira.py

@@ -31,6 +31,37 @@ def __del__(self):
         except AttributeError:
             pass
 
+    def _get_issues(self, filters: str) -> list[Issue]:
+        filters = f"{filters} AND resolution IS EMPTY"
+        data = self.client.jql(filters)
+        issues = data["issues"]
+        while len(issues) < data["total"]:
+            data = self.client.jql(filters, start=len(issues))
+            issues.extend(data["issues"])
+        return [self._to_issue(issue) for issue in issues]
+
+    def get_assigned(self, username: str = "") -> list[Issue] | None:
+        """
+        Get assigned Jira issues
+        """
+        username = username or self.client.username
+        try:
+            return self._get_issues(f"assignee = {username}")
+        except (ApiError, RequestException) as exc:
+            logging.error("Jira: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
+
+    def get_created(self, username: str = "") -> list[Issue] | None:
+        """
+        Get created Jira issues
+        """
+        username = username or self.client.username
+        try:
+            return self._get_issues(f"reporter = {username}")
+        except (ApiError, RequestException) as exc:
+            logging.error("Jira: %s: get_created(%s): %s", self.url, username, exc)
+        return None
+
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         """
         Get Jira ticket

services/launchpad.py

@@ -30,6 +30,39 @@ def __init__(self, url: str, creds: dict):
         self.issue_api_url = "https://api.launchpad.net/1.0/{repo}/+bug/{issue}"
         self.issue_web_url = f"{self.url}/{{repo}}/+bug/{{issue}}"
         self.tag = "lp"
+        self.username = creds.get("username", "")
+
+    # NOTE: Filter closed and PR's
+    def _get_issues(self, filters: dict) -> list[Issue]:
+        params = {"ws.op": "searchTasks"} | filters
+        got = self.session.get("https://api.launchpad.net/1.0/bugs", params=params)
+        got.raise_for_status()
+        entries = got.json()["entries"]
+        return [self._to_issue(entry) for entry in entries]
+
+    def get_assigned(self, username: str = "") -> list[Issue] | None:
+        """
+        Get assigned Launchpad issues
+        """
+        username = username or self.username
+        url = f"https://api.launchpad.net/1.0/~{username}"
+        try:
+            return self._get_issues({"assignee": url})
+        except RequestException as exc:
+            logging.error("Launchpad: get_assigned(%s): %s", username, exc)
+        return None
+
+    def get_created(self, username: str = "") -> list[Issue] | None:
+        """
+        Get created Launchpad issues
+        """
+        username = username or self.username
+        url = f"https://api.launchpad.net/1.0/~{username}"
+        try:
+            return self._get_issues({"bug_reporter": url})
+        except RequestException as exc:
+            logging.error("Launchpad: get_created(%s): %s", username, exc)
+        return None
 
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         if not kwargs.get("repo"):