"""Contains rules that reference GitHub."""
from __future__ import annotations
from datetime import datetime, timezone
from os import getenv
from typing import TYPE_CHECKING, cast
from attrs import define
from github3 import GitHub
from github3 import login as gh_login
from ..caching import parallel
from ..consts import EnvironmentVariable
from ..util import require_env
from . import DoResolveRule, ResolutionValueRule
if TYPE_CHECKING: # pragma: no cover
from concurrent.futures import Future
from typing import Any, Optional
from github3.issues import Issue
from github3.pulls import PullRequest
from ..market import Market
[docs]def unauth_login() -> GitHub:
"""Return an unauthorized login to GitHub."""
return GitHub()
@require_env(EnvironmentVariable.GithubAccessToken, EnvironmentVariable.GithubUsername)
def login() -> GitHub:
"""Return an authorized login to GitHub."""
return gh_login(username=getenv('GithubUsername'), token=getenv('GithubAccessToken'))
[docs]@define(slots=False)
class GitHubIssueMixin:
"""Mixin class to represent a GitHub issue."""
owner: str
repo: str
number: int
[docs] def f_issue(self) -> Future[Issue]:
"""Return a Future object which resolves to the relevant Issue object."""
return parallel(login().issue, self.owner, self.repo, self.number)
[docs] def f_pr(self) -> Future[PullRequest]:
"""Return a Future object which resolves to the relevant PullRequest object."""
return parallel(login().pull_request, self.owner, self.repo, self.number)
[docs]@define(slots=False)
class ResolveWithPR(DoResolveRule, GitHubIssueMixin):
"""Return True if the specified PR was merged in the past."""
@require_env(EnvironmentVariable.GithubAccessToken, EnvironmentVariable.GithubUsername)
def _value(self, market: Market) -> bool:
"""Return True if the issue is closed or the PR is merged, otherwise False."""
f_issue = self.f_issue()
f_pr = self.f_pr()
issue = f_issue.result()
if issue.state != 'open':
return True
pr = f_pr.result()
return pr is not None and pr.merged
[docs] def _explain_abstract(self, indent: int = 0, **kwargs: Any) -> str:
return f"{' ' * indent}- If the GitHub PR {self.owner}/{self.repo}#{self.number} was merged in the past.\n"
[docs] def _explain_specific(self, market: Market, indent: int = 0, sig_figs: int = 4) -> str:
f_issue = self.f_issue()
f_pr = self.f_pr()
ret = f"{' ' * indent}- If either of the conditions below are True (-> {self.value(market)})\n"
indent += 1
issue = f_issue.result()
ret += f"{' ' * indent}- If the state of the pull request is not open (-> {issue.state})\n"
pr = f_pr.result()
ret += f"{' ' * indent}- If the pull request is merged (-> {pr is not None and pr.merged})\n"
return ret
[docs]@define(slots=False)
class ResolveToPR(ResolutionValueRule, GitHubIssueMixin):
"""Resolve to True if the PR is merged, otherwise False."""
[docs] def _value(self, market: Market) -> bool:
pr = self.f_pr().result()
return pr is not None and pr.merged
[docs] def _explain_abstract(self, indent: int = 0, **kwargs: Any) -> str:
ret = f"{' ' * indent}- Resolves based on GitHub PR {self.owner}/{self.repo}#{self.number}\n"
indent += 1
ret += f"{' ' * indent}- If the PR is merged, resolve to YES.\n"
ret += f"{' ' * indent}- Otherwise, resolve to NO.\n"
return ret
[docs] def _explain_specific(self, market: Market, indent: int = 0, sig_figs: int = 4) -> str:
pr = self.f_pr().result()
if pr is None:
merge_time = None
else:
merge_time = pr.merged_at
return (f"{' ' * indent}- Is the pull request is merged? (-> {merge_time or 'Not yet merged'} -> "
f"{merge_time is not None})\n")
[docs]@define(slots=False)
class ResolveToPRDelta(ResolutionValueRule, GitHubIssueMixin):
"""Resolve to the fractional number of days between start and merged date or, if not merged, MAX."""
start: datetime
[docs] def _value(self, market: Market) -> float:
pr = self.f_pr().result()
if pr is None or pr.merged_at is None:
return cast(float, market.market.max)
delta = cast(datetime, pr.merged_at) - self.start.replace(tzinfo=timezone.utc)
return delta.days + (delta.seconds / (24 * 60 * 60))
[docs] def _explain_abstract(self, indent: int = 0, max_: Optional[float] = None, **kwargs: Any) -> str:
ret = f"{' ' * indent}- Resolves based on GitHub PR {self.owner}/{self.repo}#{self.number}\n"
indent += 1
ret += (f"{' ' * indent}- If the PR is merged, resolve to the number of days between {self.start} and the "
"resolution time.\n")
ret += f"{' ' * indent}- Otherwise, resolve to MAX"
if max_ is not None:
ret += f" ({max_})"
ret += ".\n"
return ret
[docs] def _explain_specific(self, market: Market, indent: int = 0, sig_figs: int = 4) -> str:
pr = self.f_pr().result()
if pr is None:
merge_time = None
else:
merge_time = pr.merged_at
return (f"{' ' * indent}- How long after {self.start} was the pull request is merged? (-> "
f"{merge_time or 'Not yet merged'} -> {self.value(market)})\n")