diff --git a/services/__init__.py b/services/__init__.py index 8e80860..d4a0043 100644 --- a/services/__init__.py +++ b/services/__init__.py @@ -166,6 +166,9 @@ def __exit__(self, exc_type, exc_value, traceback): def __repr__(self) -> str: return f"{self.__class__.__name__}(url='{self.url}')" + def __hash__(self) -> int: + return hash(self.url) + def _not_found(self, url: str, tag: str) -> Issue: now = datetime.now(tz=utc) return Issue( diff --git a/services/gitlab.py b/services/gitlab.py index 722b65d..68a95f4 100644 --- a/services/gitlab.py +++ b/services/gitlab.py @@ -45,21 +45,12 @@ def __del__(self): except (AttributeError, GitlabError): pass - def get_user_issues( # pylint: disable=too-many-arguments - self, - username: str = "", - assigned: bool = False, - created: bool = False, - involved: bool = True, - pull_requests: bool = False, - state: str = "opened", - **_, + def get_assigned( + self, username: str = "", pull_requests: bool = False, state: str = "Open" ) -> list[Issue] | None: """ - Get user issues + Get assigned issues """ - if involved: - assigned = created = True filters = { "all": True, # No pagination "state": state, @@ -70,38 +61,71 @@ def get_user_issues( # pylint: disable=too-many-arguments user = self.client.users.list(username=username)[0] # type: ignore else: user = self.client.user - if assigned: - if pull_requests: - issues = list( - self.client.mergerequests.list(assignee_id=user.id, **filters) - ) - else: - issues = list( - self.client.issues.list(assignee_id=user.id, **filters) - ) - found_ids = {i.iid for i in issues} - if created: - if pull_requests: - issues.extend( - issue - for issue in self.client.mergerequests.list( - author=user.id, **filters - ) - if issue.iid not in found_ids - ) - else: - issues.extend( - issue - for issue in self.client.issues.list(author=user.id, **filters) - if issue.iid not in found_ids - ) + if pull_requests: + issues = list( + self.client.mergerequests.list(assignee_id=user.id, **filters) + ) + else: + issues = list(self.client.issues.list(assignee_id=user.id, **filters)) except (GitlabError, RequestException) as exc: - logging.error( - "Gitlab: %s: get_user_issues(%s): %s", self.url, username, exc - ) + logging.error("Gitlab: %s: get_assigned(%s): %s", self.url, username, exc) + return None + return [self._to_issue(issue) for issue in issues] + + def get_created( + self, username: str = "", pull_requests: bool = False, state: str = "Open" + ) -> list[Issue] | None: + """ + Get created issues + """ + filters = { + "all": True, # No pagination + "state": state, + } + issues: list[Any] = [] + try: + if username: + user = self.client.users.list(username=username)[0] # type: ignore + else: + user = self.client.user + if pull_requests: + issues = list(self.client.mergerequests.list(author=user.id, **filters)) + else: + issues = list(self.client.issues.list(author=user.id, **filters)) + except (GitlabError, RequestException) as exc: + logging.error("Gitlab: %s: get_created(%s): %s", self.url, username, exc) return None return [self._to_issue(issue) for issue in issues] + def get_user_issues( # pylint: disable=too-many-arguments + self, + username: str = "", + assigned: bool = False, + created: bool = False, + involved: bool = True, + state: str = "opened", + **_, + ) -> list[Issue] | None: + """ + Get user issues + """ + if involved: + assigned = created = True + issues: list[Issue] = [] + if assigned: + for is_pr in (False, True): + more = self.get_assigned(username, pull_requests=is_pr, state=state) + if more is None: + return None + issues.extend(more) + if created: + for is_pr in (False, True): + more = self.get_created(username, pull_requests=is_pr, state=state) + if more is None: + return None + issues.extend(more) + return list(set(issues)) + def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None: """ Get Gitlab issue diff --git a/services/jira.py b/services/jira.py index c2f6650..3abe31d 100644 --- a/services/jira.py +++ b/services/jira.py @@ -42,6 +42,18 @@ def _get_issues(self, filters: str) -> list[dict]: issues.extend(data["issues"]) return issues + def get_assigned(self, username: str = "") -> list[Issue] | None: + """ + Get assigned issues + """ + return self.get_user_issues(username, assigned=True, involved=False) + + def get_created(self, username: str = "") -> list[Issue] | None: + """ + Get created issues + """ + return self.get_user_issues(username, created=True, involved=False) + def get_user_issues( # pylint: disable=too-many-arguments self, username: str = "", diff --git a/services/launchpad.py b/services/launchpad.py index 55f7cb0..d0d0ce3 100644 --- a/services/launchpad.py +++ b/services/launchpad.py @@ -32,6 +32,48 @@ def __init__(self, url: str, creds: dict): self.tag = "lp" self.username = creds.get("username", "") + def get_assigned(self, username: str = "") -> list[Issue] | None: + """ + Get assigned issues + """ + username = username or self.username + user_url = f"https://api.launchpad.net/1.0/~{username}" + params = { + "ws.op": "searchTasks", + } + try: + got = self.session.get( + "https://api.launchpad.net/1.0/bugs", + params=params | {"assignee": user_url}, + ) + got.raise_for_status() + issues = got.json()["entries"] + except RequestException as exc: + logging.error("Launchpad: get_assigned(%s): %s", username, exc) + return None + return [self._to_issue(issue) for issue in issues] + + def get_created(self, username: str = "") -> list[Issue] | None: + """ + Get created issues + """ + username = username or self.username + user_url = f"https://api.launchpad.net/1.0/~{username}" + params = { + "ws.op": "searchTasks", + } + try: + got = self.session.get( + "https://api.launchpad.net/1.0/bugs", + params=params | {"bug_reporter": user_url}, + ) + got.raise_for_status() + issues = got.json()["entries"] + except RequestException as exc: + logging.error("Launchpad: get_created(%s): %s", username, exc) + return None + return [self._to_issue(issue) for issue in issues] + def get_user_issues( # pylint: disable=too-many-arguments self, username: str = "", @@ -45,36 +87,18 @@ def get_user_issues( # pylint: disable=too-many-arguments """ if involved: assigned = created = True - username = username or self.username - user_url = f"https://api.launchpad.net/1.0/~{username}" - params = { - "ws.op": "searchTasks", - } - issues = [] - try: - if assigned: - got = self.session.get( - "https://api.launchpad.net/1.0/bugs", - params=params | {"assignee": user_url}, - ) - got.raise_for_status() - issues = got.json()["entries"] - found_ids = {i["web_link"] for i in issues} - if created: - got = self.session.get( - "https://api.launchpad.net/1.0/bugs", - params=params | {"bug_reporter": user_url}, - ) - got.raise_for_status() - issues.extend( - issue - for issue in got.json()["entries"] - if issue["web_link"] not in found_ids - ) - except RequestException as exc: - logging.error("Launchpad: get_user_issues(%s): %s", username, exc) - return None - return [self._to_issue(issue) for issue in issues] + issues: list[Issue] = [] + if assigned: + more = self.get_assigned(username) + if more is None: + return None + issues.extend(more) + if created: + more = self.get_created(username) + if more is None: + return None + issues.extend(more) + return list(set(issues)) def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None: if not kwargs.get("repo"): diff --git a/services/pagure.py b/services/pagure.py index 10d1036..f4ccd5f 100644 --- a/services/pagure.py +++ b/services/pagure.py @@ -73,13 +73,52 @@ def _get_pullrequests( url, params=params, key="requests", next_key="pagination" ) + def get_assigned( + self, username: str = "", pull_requests: bool = False, state: str = "Open" + ) -> list[Issue] | None: + """ + Get assigned issues + """ + username = username or self.username + filters = { + "status": state, + } + try: + if pull_requests: + issues = self._get_pullrequests(username, created=False, **filters) + else: + issues = self._get_issues(username, assignee=1, author=0, **filters) + except RequestException as exc: + logging.error("Pagure: %s: get_assigned(%s): %s", self.url, username, exc) + return None + return [self._to_issue(issue, is_pr=pull_requests) for issue in issues] + + def get_created( + self, username: str = "", pull_requests: bool = False, state: str = "Open" + ) -> list[Issue] | None: + """ + Get created issues + """ + username = username or self.username + filters = { + "status": state, + } + try: + if pull_requests: + issues = self._get_pullrequests(username, created=True, **filters) + else: + issues = self._get_issues(username, assignee=0, author=1, **filters) + except RequestException as exc: + logging.error("Pagure: %s: get_created(%s): %s", self.url, username, exc) + return None + return [self._to_issue(issue, is_pr=pull_requests) for issue in issues] + def get_user_issues( # pylint: disable=too-many-arguments self, username: str = "", assigned: bool = False, created: bool = False, involved: bool = True, - pull_requests: bool = False, state: str = "Open", **_, ) -> list[Issue] | None: @@ -88,42 +127,20 @@ def get_user_issues( # pylint: disable=too-many-arguments """ if involved: assigned = created = True - username = username or self.username - filters = { - "status": state, - } - issues = [] - try: - if pull_requests: - if assigned: - issues = self._get_pullrequests(username, created=False, **filters) - found_ids = {i["id"] for i in issues} - if created: - issues.extend( - issue - for issue in self._get_pullrequests( - username, created=True, **filters - ) - if issue["id"] not in found_ids - ) - else: - if assigned: - issues = self._get_issues(username, assignee=1, author=0, **filters) - found_ids = {i["id"] for i in issues} - if created: - issues.extend( - issue - for issue in self._get_issues( - username, assignee=0, author=1, **filters - ) - if issue["id"] not in found_ids - ) - except RequestException as exc: - logging.error( - "Pagure: %s: get_user_issues(%s): %s", self.url, username, exc - ) - return None - return [self._to_issue(issue, is_pr=pull_requests) for issue in issues] + issues: list[Issue] = [] + if assigned: + for is_pr in (False, True): + more = self.get_assigned(username, pull_requests=is_pr, state=state) + if more is None: + return None + issues.extend(more) + if created: + for is_pr in (False, True): + more = self.get_created(username, pull_requests=is_pr, state=state) + if more is None: + return None + issues.extend(more) + return list(set(issues)) def _to_issue(self, info: Any, **kwargs) -> Issue: repo = kwargs.get("repo", "") or info["project"]["fullname"] diff --git a/services/redmine.py b/services/redmine.py index 8fa36bd..3634bc5 100644 --- a/services/redmine.py +++ b/services/redmine.py @@ -36,6 +36,30 @@ def __del__(self): except AttributeError: pass + def get_assigned(self, username: str = "me") -> list[Issue] | None: + """ + Get assigned issues + """ + try: + user = self.client.user.get(username) + issues = self.client.issue.filter(assigned_to_id=user.id) + except (BaseRedmineError, RequestException) as exc: + logging.error("Redmine: %s: get_assigned(%s): %s", self.url, username, exc) + return None + return [self._to_issue(issue) for issue in issues] + + def get_created(self, username: str = "me") -> list[Issue] | None: + """ + Get created issues + """ + try: + user = self.client.user.get(username) + issues = self.client.issue.filter(author=user.id) + except (BaseRedmineError, RequestException) as exc: + logging.error("Redmine: %s: get_created(%s): %s", self.url, username, exc) + return None + return [self._to_issue(issue) for issue in issues] + def get_user_issues( # pylint: disable=too-many-arguments self, username: str = "me", @@ -49,24 +73,18 @@ def get_user_issues( # pylint: disable=too-many-arguments """ if involved: assigned = created = True - issues = [] - try: - user = self.client.user.get(username) - if assigned: - issues = list(self.client.issue.filter(assigned_to_id=user.id)) - found_ids = {i.id for i in issues} - if created: - issues.extend( - issue - for issue in self.client.issue.filter(author_id=user.id) - if issue.id not in found_ids - ) - except (BaseRedmineError, RequestException) as exc: - logging.error( - "Redmine: %s: get_user_issues(%s): %s", self.url, username, exc - ) - return None - return [self._to_issue(issue) for issue in issues] + issues: list[Issue] = [] + if assigned: + more = self.get_assigned(username) + if more is None: + return None + issues.extend(more) + if created: + more = self.get_created(username) + if more is None: + return None + issues.extend(more) + return list(set(issues)) def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None: """