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 42064ee
Show file tree
Hide file tree
Showing 6 changed files with 223 additions and 125 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
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
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(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"]
Expand Down
Loading

0 comments on commit 42064ee

Please sign in to comment.