From c7c739680acd66bbe74057b474a253e371c57f26 Mon Sep 17 00:00:00 2001 From: Ricardo Branco Date: Thu, 19 Oct 2023 13:04:43 +0200 Subject: [PATCH] Introduce get_assigned() & get_created() --- services/__init__.py | 3 ++ services/pagure.py | 91 ++++++++++++++++++++++++++------------------ services/redmine.py | 54 +++++++++++++++++--------- 3 files changed, 93 insertions(+), 55 deletions(-) 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/pagure.py b/services/pagure.py index 10d1036..4d5d112 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 = "me", 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 = "me", 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(self._to_issue(issue, is_pr=is_pr) for issue in 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(self._to_issue(issue, is_pr=is_pr) for issue in 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: """