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 6a88e5b
Show file tree
Hide file tree
Showing 4 changed files with 147 additions and 85 deletions.
3 changes: 3 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 Down
84 changes: 54 additions & 30 deletions services/launchpad.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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 = "",
Expand All @@ -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"):
Expand Down
91 changes: 54 additions & 37 deletions services/pagure.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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"]
Expand Down
54 changes: 36 additions & 18 deletions services/redmine.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand All @@ -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:
"""
Expand Down

0 comments on commit 6a88e5b

Please sign in to comment.