-
Notifications
You must be signed in to change notification settings - Fork 27
/
gh_issue.py
84 lines (63 loc) · 2.5 KB
/
gh_issue.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
"""Functionality for working with GitHub issues."""
import logging
import typing as tp
import github
LOGGER = logging.getLogger(__name__)
class GHIssue:
"""GitHub issue."""
TOKEN: tp.ClassVar[tp.Optional[str]] = None
issue_cache: tp.ClassVar[tp.Dict[str, str]] = {}
_github_instance: tp.ClassVar[tp.Optional[github.Github]] = None
_github_instance_error: tp.ClassVar[bool] = False
@classmethod
def _get_github(cls) -> tp.Optional[github.Github]:
"""Get GitHub instance."""
if cls._github_instance is not None:
return cls._github_instance
if cls._github_instance_error:
return None
try:
# Max 60 req/hr without token
cls._github_instance = (
github.Github(login_or_token=cls.TOKEN) if cls.TOKEN else github.Github()
)
except Exception:
# pylint: disable=broad-except
LOGGER.exception("Failed to get GitHub instance")
cls._github_instance_error = True
return None
return cls._github_instance
def __init__(self, number: int, repo: str) -> None:
self.number = number
self.repo = repo
@property
def github(self) -> tp.Optional[github.Github]:
return self._get_github()
@property
def url(self) -> str:
return f"https://github.com/{self.repo}/issues/{self.number}"
def get_state(self) -> tp.Optional[str]:
"""Get issue state."""
if not self.github:
LOGGER.error("Failed to get GitHub instance")
return None
identifier = f"{self.repo}#{self.number}"
cached_state = self.issue_cache.get(identifier)
if cached_state is None:
assert self.github # for pytype
try:
cached_state = self.github.get_repo(self.repo).get_issue(self.number).state.lower()
except github.UnknownObjectException:
LOGGER.exception("Unknown issue '%s'", identifier)
cached_state = "unknown"
except Exception:
# pylint: disable=broad-except
LOGGER.exception("Failed to get issue '%s'", identifier)
cached_state = "get_state_failure"
self.issue_cache[identifier] = cached_state
return cached_state
def is_closed(self) -> bool:
"""Check if issue is closed."""
return self.get_state() == "closed"
def __repr__(self) -> str:
return f"<GHIssue: {self.repo}#{self.number}>"