Skip to content

Commit

Permalink
Introduce get_assigned() & get_created()
Browse files Browse the repository at this point in the history
  • Loading branch information
@ricardobranco777
ricardobranco777 committed Oct 19, 2023
1 parent 0aa8f5a commit a9c4769
Show file tree
Hide file tree
Showing 8 changed files with 313 additions and 155 deletions.
15 changes: 15 additions & 0 deletions services/__init__.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand All @@ -186,6 +189,18 @@ def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
"""
raise NotImplementedError(f"{self.__class__.__name__}: get_issue()")

def get_assigned(self, **_) -> list[Issue] | None:
"""
Get assigned issues
"""
return []

def get_created(self, **_) -> list[Issue] | None:
"""
Get created issues
"""
return []

def get_user_issues(self, **_) -> list[Issue] | None:
"""
Get user issues
Expand Down
67 changes: 46 additions & 21 deletions services/bugzilla.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,40 @@ def __del__(self):
except (AttributeError, BugzillaError):
pass

def get_assigned(
self, username: str = "", closed: bool = False, **_
) -> list[Issue] | None:
"""
Get assigned issues
"""
username = username or self.client.user
try:
user = self.client.getuser(username)
issues = self.client.query({"assigned_to": user.email})
except (AttributeError, BugzillaError, RequestException) as exc:
logging.error("Bugzilla: %s: get_assigned(%s): %s", self.url, username, exc)
return None
if not closed:
issues = [issue for issue in issues if issue.is_open]
return [self._to_issue(issue) for issue in issues]

def get_created(
self, username: str = "", closed: bool = False, **_
) -> list[Issue] | None:
"""
Get created issues
"""
username = username or self.client.user
try:
user = self.client.getuser(username)
issues = self.client.query({"reporter": user.email})
except (AttributeError, BugzillaError, RequestException) as exc:
logging.error("Bugzilla: %s: get_created(%s): %s", self.url, username, exc)
return None
if not closed:
issues = [issue for issue in issues if issue.is_open]
return [self._to_issue(issue) for issue in issues]

def get_user_issues( # pylint: disable=too-many-arguments
self,
username: str = "",
Expand All @@ -54,27 +88,18 @@ def get_user_issues( # pylint: disable=too-many-arguments
"""
if involved:
assigned = created = True
username = username or self.client.user
issues = []
try:
user = self.client.getuser(username)
if assigned:
issues = self.client.query({"assigned_to": user.email})
found_ids = {i.id for i in issues}
if created:
issues.extend(
issue
for issue in self.client.query({"reporter": user.email})
if issue.id not in found_ids
)
except (AttributeError, BugzillaError, RequestException) as exc:
logging.error(
"Bugzilla: %s: get_user_issues(%s): %s", self.url, username, exc
)
return None
if not closed:
issues = [issue for issue in issues if issue.is_open]
return [self._to_issue(issue) for issue in issues]
issues: list[Issue] = []
if assigned:
more = self.get_assigned(username, closed=closed)
if more is None:
return None
issues.extend(more)
if created:
more = self.get_created(username, closed=closed)
if more is None:
return None
issues.extend(more)
return list(set(issues))

def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
"""
Expand Down
41 changes: 32 additions & 9 deletions services/github.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -36,37 +36,60 @@ def __del__(self):
except (AttributeError, GithubException):
pass

def get_assigned(
self, username: str = "", state: str = "open", **_
) -> list[Issue] | None:
"""
Get assigned issues
"""
return self.get_user_issues(
username, assigned=True, involved=False, state=state
)

def get_created(
self, username: str = "", state: str = "open", **_
) -> list[Issue] | None:
"""
Get created issues
"""
return self.get_user_issues(username, created=True, involved=False, state=state)

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:
"""
Get user issues
"""
issue_type = "pr" if pull_requests else "issue"
filters = f"state:{state} type:{issue_type} "
base_filters = f"state:{state} "
if involved:
filters += "involves"
base_filters += "involves"
elif assigned:
filters += "assignee"
base_filters += "assignee"
elif created:
filters += "author"
base_filters += "author"
issues: list[Issue] = []
try:
user = (
self.client.get_user(username) if username else self.client.get_user()
)
filters = f"{filters}:{user.login}"
issues = self.client.search_issues(filters)
base_filters = f"{base_filters}:{user.login}"
for issue_type in ("pr", "issue"):
filters = f"{base_filters} type:{issue_type}"
more = self.client.search_issues(filters)
issues.extend(
self._to_issue(issue, is_pr=bool(issue_type == "pr"))
for issue in more
)
except (GithubException, RequestException) as exc:
logging.error("Github: get_user_issues(%s): %s", username, exc)
return None
return [self._to_issue(issue, is_pr=pull_requests) for issue in issues]
return issues

def get_issue(self, issue_id: str = "", **kwargs) -> Issue | None:
"""
Expand Down
104 changes: 64 additions & 40 deletions services/gitlab.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand All @@ -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
Expand Down
12 changes: 12 additions & 0 deletions services/jira.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 = "",
Expand Down
Loading

0 comments on commit a9c4769

Please sign in to comment.