Introduce get_created() & get_assigned()

services/bugzilla.py

@@ -39,6 +39,33 @@ 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
+        """
+        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/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,43 @@ def __del__(self):
         except (AttributeError, GithubException):
             pass
 
+    def _get_issues(
+        self, filters: str, is_pr: bool, state: str = "open"
+    ) -> list[Issue]:
+        issue_type = "pr" if is_pr else "issue"
+        filters = f"{filters} type:{issue_type} state:{state}"
+        return [self._to_issue(issue) for issue in self.client.search_issues(filters)]
+
+    def get_assigned(
+        self, username: str | None = None, is_pr: bool = False
+    ) -> 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}", 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
+    ) -> 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}", 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 +98,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
+    ) -> 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)
+            return self._get_issues(assignee_id=user.id)
+        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
+    ) -> 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)
+            return self._get_issues(author=user.id)
+        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/launchpad.py

@@ -31,6 +31,36 @@ def __init__(self, url: str, creds: dict):
         self.issue_web_url = f"{self.url}/{{repo}}/+bug/{{issue}}"
         self.tag = "lp"
 
+    # 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 | None = None) -> list[Issue] | None:
+        """
+        Get assigned Launchpad issues
+        """
+        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 | None = None) -> list[Issue] | None:
+        """
+        Get created Launchpad issues
+        """
+        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"):
             try:

services/pagure.py

@@ -2,8 +2,11 @@
 Pagure
 """
 
+import logging
 from typing import Any
 
+from requests.exceptions import RequestException
+
 from utils import utc_date
 from . import Generic, Issue, status
 
@@ -21,8 +24,72 @@ def __init__(self, url: str, creds: dict):
         self.pr_api_url = f"{self.url}/api/0/{{repo}}/pull-request/{{issue}}"
         self.pr_web_url = f"{self.url}/{{repo}}/pull-request/{{issue}}"
 
+    def _get_issues(self, username: str, **params) -> list[Issue]:
+        if params["assignee"]:
+            key = "issues_assigned"
+            next_key = "pagination_issues_assigned"
+        else:
+            key = "issues_created"
+            next_key = "pagination_issues_created"
+        got = self.session.get(
+            f"{self.url}/api/0/user/{username}/issues", params=params
+        )
+        got.raise_for_status()
+        data = got.json()
+        entries = data[key]
+        while data[next_key]["next"]:
+            got = self.session.get(data[next_key]["next"], params=params)
+            got.raise_for_status()
+            data = got.json()
+            entries.extend(data[key])
+        return [self._to_issue(entry) for entry in entries]
+
+    def _get_pullrequests(
+        self, username: str, created: bool = False, state: str = "Open"
+    ) -> list[Issue]:
+        pr_type = "filed" if created else "actionable"
+        params = {
+            "status": state,
+        }
+        got = self.session.get(
+            f"{self.url}/api/0/user/{username}/requests/{pr_type}", params=params
+        )
+        got.raise_for_status()
+        data = got.json()
+        entries = data["requests"]
+        while data["pagination"]["next"]:
+            got = self.session.get(data["pagination"]["next"], params=params)
+            got.raise_for_status()
+            data = got.json()
+            entries.extend(data["requests"])
+        return [self._to_issue(entry) for entry in entries]
+
+    def get_assigned(self, username: str, is_pr: bool = False) -> list[Issue] | None:
+        """
+        Get assigned Pagure issues
+        """
+        try:
+            if is_pr:
+                return self._get_pullrequests(username)
+            return self._get_issues(username, assignee=1, author=0)
+        except RequestException as exc:
+            logging.error("Pagure: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
+
+    def get_created(self, username: str, is_pr: bool = False) -> list[Issue] | None:
+        """
+        Get assigned Pagure issues
+        """
+        try:
+            if is_pr:
+                return self._get_pullrequests(username, created=True)
+            return self._get_issues(username, assignee=0, author=1)
+        except RequestException as exc:
+            logging.error("Pagure: %s: get_created(%s): %s", self.url, username, exc)
+        return None
+
     def _to_issue(self, info: Any, **kwargs) -> Issue:
-        repo = kwargs.pop("repo")
+        repo = kwargs.get("repo", "") or info["project"]["fullname"]
         is_pr = kwargs.pop("is_pr")
         mark = "!" if is_pr else "#"
         return Issue(

services/redmine.py

@@ -36,6 +36,31 @@ def __del__(self):
         except AttributeError:
             pass
 
+    def _get_issues(self, **filters) -> list[Issue]:
+        return [self._to_issue(issue) for issue in self.client.issue.filter(**filters)]
+
+    def get_assigned(self, username: str = "me") -> list[Issue] | None:
+        """
+        Get assigned Redmine tickets
+        """
+        try:
+            user = self.client.user.get(username)
+            return self._get_issues(assigned_to_id=user.id)
+        except (BaseRedmineError, RequestException) as exc:
+            logging.error("Redmine: %s: get_assigned(%s): %s", self.url, username, exc)
+        return None
+
+    def get_created(self, username: str = "me") -> list[Issue] | None:
+        """
+        Get created Redmine tickets
+        """
+        try:
+            user = self.client.user.get(username)
+            return self._get_issues(author_id=user.id)
+        except (BaseRedmineError, RequestException) as exc:
+            logging.error("Redmine: %s: get_created(%s): %s", self.url, username, exc)
+        return None
+
     def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
         """
         Get Redmine ticket