Skip to content

Commit

Permalink
Refactor into classes
Browse files Browse the repository at this point in the history
  • Loading branch information
@ricardobranco777
ricardobranco777 committed Sep 8, 2023
1 parent 45e0e28 commit ef09ff5
Show file tree
Hide file tree
Showing 2 changed files with 177 additions and 53 deletions.
2 changes: 1 addition & 1 deletion Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,4 +9,4 @@ flake8:

.PHONY: pylint
pylint:
@pylint $(FILES)
@pylint --disable=line-too-long $(FILES)
228 changes: 176 additions & 52 deletions bugme.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,12 @@
Bugme
"""

import logging
import os
import json
import sys
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlparse

from dateutil import parser
from pytz import utc
Expand All @@ -30,77 +33,198 @@ def dateit(date, time_format: str = "%a %b %d %H:%M:%S %Z %Y") -> str:
return date.astimezone().strftime(time_format)


def do_bugzilla(url: str, bugs: list, creds):
class RepoIssue: # pylint: disable=too-few-public-methods
"""
Simple class to hold GitHub issue
"""
def __init__(self, repo: str, issue: str):
self.repo = repo
self.number = int(issue)


class Item: # pylint: disable=too-few-public-methods
"""
Item class
"""
def __init__(self, **kwargs):
for attr, value in kwargs.items():
setattr(self, attr, value)


class Service:
"""
Service class to abstract methods
"""
def __init__(self, url):
url = url.rstrip('/')
self.url = url if urlparse(url).scheme else f"https://{url}"

def __enter__(self):
return self

def __exit__(self, exc_type, exc_value, traceback):
if exc_type is not None:
logging.error("%s: %s: %s: %s", self.__class__.__name__, exc_type, exc_value, traceback)

def get_item(self, item: int) -> Item:
"""
This method must be overriden if get_items() isn't overriden
"""
raise NotImplementedError(f"{self.__class__.__name__}: get_item()")

def get_items(self, items: list) -> list:
"""
Multithreaded get_items()
"""
with ThreadPoolExecutor(max_workers=len(items)) as executor:
yield from executor.map(self.get_item, items)


class MyBugzilla(Service):
"""
Bugzilla
"""
if len(bugs) == 0:
return
sslverify = os.environ.get("REQUESTS_CA_BUNDLE", True)
try:
mybugz = Bugzilla(url, force_rest=True, sslverify=sslverify, **creds)
for bug in mybugz.getbugs(bugs):
print(f"bsc#{bug.id}\t{bug.status}\t\t{dateit(bug.last_change_time)}\t{bug.summary}")
mybugz.disconnect()
except BugzillaError as exc:
print(f"Bugzilla: {exc}")
def __init__(self, url: str, creds: dict):
super().__init__(url)
sslverify = os.environ.get("REQUESTS_CA_BUNDLE", True)
self.client = Bugzilla(self.url, force_rest=True, sslverify=sslverify, **creds)

def __exit__(self, exc_type, exc_value, traceback):
try:
self.client.disconnect()
except BugzillaError:
pass
super().__exit__(exc_type, exc_value, traceback)

def get_item(self, item: int) -> Item:
"""
Get Bugzilla item
"""
try:
return self._to_item(self.client.getbug(item))
except BugzillaError as exc:
logging.error("Bugzilla: %s: get_items(%d): %s", self.url, item, exc)
return None

def get_items(self, items: list) -> list:
"""
Get Bugzilla items
"""
try:
for info in self.client.getbugs(items):
yield self._to_item(info)
except BugzillaError as exc:
logging.error("Bugzilla: %s: get_items(): %s", self.url, exc)

def _to_item(self, info) -> Item:
return Item(
id=info.id,
status=info.status,
title=info.summary,
updated_at=info.last_change_time,
url=f"{self.url}/show_bug.cgi?id={info.id}",
extra=info.__dict__,
)


def do_github(issues: list, creds: dict):
class MyGithub(Service):
"""
Github
"""
if len(issues) == 0:
return
auth = Auth.Token(**creds)
mygh = Github(auth=auth)
for issue in issues:
def __init__(self, creds: dict):
super().__init__("https://github.com")
auth = Auth.Token(**creds)
self.client = Github(auth=auth)

def get_item(self, issue: RepoIssue) -> Item:
"""
Get Github issue
"""
try:
info = mygh.get_repo(issue.repo, lazy=True).get_issue(issue.number)
print(f"gh#{info.number}\t{info.state}\t\t{dateit(info.last_modified)}\t{info.title}")
info = self.client.get_repo(issue.repo, lazy=True).get_issue(issue.number)
except GithubException as exc:
print(f"gh#{issue.repo}#{issue.number}: {exc}", file=sys.stderr)
logging.error("Github: get_issue(%d): %s", issue, exc)
return None
return self._to_item(info, issue)

def _to_item(self, info, issue) -> Item:
return Item(
id=info.number,
status=info.state,
title=info.title,
updated_at=info.last_modified,
url=f"{self.url}/{issue.repo}/issues/{issue.number}",
extra=info.__dict__["_rawData"],
)


def do_gitlab(url: str, issues: list, creds: dict):
class MyGitlab(Service):
"""
Gitlab
"""
if len(issues) == 0:
return
ssl_verify = os.getenv("REQUESTS_CA_BUNDLE") if url else True
with Gitlab(url=url, ssl_verify=ssl_verify, retry_transient_errors=False, **creds) as mygl:
for issue in issues:
try:
info = mygl.projects.get(issue.repo, lazy=True).issues.get(issue.number)
print(f"gl#{info.iid}\t{info.state}\t\t{dateit(info.updated_at)}\t{info.title}")
except GitlabError as exc:
print(f"gl#{issue.repo}#{issue.number}: {exc}", file=sys.stderr)
def __init__(self, url: str, creds: dict):
super().__init__(url)
ssl_verify = os.getenv("REQUESTS_CA_BUNDLE") if self.url else True
self.client = Gitlab(url=self.url, ssl_verify=ssl_verify, retry_transient_errors=False, **creds)

def __exit__(self, exc_type, exc_value, traceback):
try:
self.client.__exit__(exc_type, exc_value, traceback)
except GitlabError:
pass
super().__exit__(exc_type, exc_value, traceback)

def get_item(self, issue: RepoIssue) -> Item:
"""
Get Gitlab issue
"""
try:
info = self.client.projects.get(issue.repo, lazy=True).issues.get(issue.number)
except GitlabError as exc:
logging.error("Gitlab: %s: get_issue(%d): %s", self.url, issue, exc)
return None
return self._to_item(info, issue)

def _to_item(self, info, issue) -> Item:
return Item(
id=info.iid,
status=info.state,
title=info.title,
updated_at=info.updated_at,
url=f"{self.url}/{issue.repo}/-/issues/{issue.number}",
extra=info.asdict(),
)

def do_redmine(url: str, tickets: list, creds: dict):

class MyRedmine(Service):
"""
Redmine
"""
if len(tickets) == 0:
return
redmine = Redmine(url=url, raise_attr_exception=False, **creds)
for ticket in tickets:
def __init__(self, url: str, creds: dict):
super().__init__(url)
self.client = Redmine(url=self.url, raise_attr_exception=False, **creds)

def get_item(self, item: int) -> Item:
"""
Get Redmine ticket
"""
try:
info = redmine.issue.get(ticket)
info = self.client.issue.get(item)
print(f"poo#{info.id}\t{info.status}\t{dateit(info.updated_on)}\t{info.subject}")
except BaseRedmineError as exc:
print(f"poo#{ticket}: {exc}", file=sys.stderr)

logging.error("Redmine: %s: get_issue(%d): %s", self.url, item, exc)
return None
return self._to_item(info)

class RepoIssue: # pylint: disable=too-few-public-methods
"""
Simple class to hold GitHub issue
"""

def __init__(self, repo: str, issue: str):
self.repo = repo
self.number = int(issue)
def _to_item(self, info) -> Item:
return Item(
id=info.id,
status=info.status,
title=info.subject,
updated_at=info.updated_on,
url=f"{self.url}/issues/{info.id}",
extra=info.raw(),
)


def main():
Expand Down Expand Up @@ -132,11 +256,11 @@ def main():
sys.exit(f"ERROR: {CREDENTIALS_FILE} has insecure permissions")
creds = json.load(file)

do_bugzilla("https://bugzilla.suse.com", bsc_list, creds["bugzilla.suse.com"])
do_github(gh_list, creds["github.com"])
do_gitlab(None, gl_list, creds["gitlab.com"])
do_gitlab("https://gitlab.suse.de", gsd_list, creds["gitlab.suse.de"])
do_redmine("https://progress.opensuse.org", poo_list, creds["progress.opensuse.org"])
# do_bugzilla("https://bugzilla.suse.com", bsc_list, creds["bugzilla.suse.com"])
# do_github(gh_list, creds["github.com"])
# do_gitlab(None, gl_list, creds["gitlab.com"])
# do_gitlab("https://gitlab.suse.de", gsd_list, creds["gitlab.suse.de"])
# do_redmine("https://progress.opensuse.org", poo_list, creds["progress.opensuse.org"])


if __name__ == "__main__":
Expand Down

0 comments on commit ef09ff5

Please sign in to comment.