""" Registry for GitHub event handlers. """ import dataclasses import logging import sys import threading import typing as t import deprecated import requests import urllib3 # type: ignore[import] from nr.functional import coalesce from . import __version__ from .token import InstallationTokenSupplier, JwtSupplier, TokenInfo T = t.TypeVar('T') logger = logging.getLogger(__name__) user_agent = f'python/{sys.version.split()[0]} github-bot-api/{__version__}' if t.TYPE_CHECKING: import github @dataclasses.dataclass class GithubClientSettings: """ Settings for constructing a #github.Github client object. """ base_url: t.Optional[str] = None user_agent: t.Optional[str] = None timeout: t.Optional[float] = None per_page: t.Optional[int] = None verify: t.Optional[bool] = None retry: t.Optional[urllib3.Retry] = None def update(self, other: 'GithubClientSettings') -> 'GithubClientSettings': result = GithubClientSettings() for field in dataclasses.fields(self): value = getattr(other, field.name) if value is None: value = getattr(self, field.name) setattr(result, field.name, value) return result def make_client(self, login_or_token: t.Optional[str] = None, jwt: t.Optional[str] = None) -> 'github.Github': import github, github.MainClass return github.Github( login_or_token=login_or_token, jwt=jwt, base_url=self.base_url or github.MainClass.DEFAULT_BASE_URL, # type: ignore[attr-defined] user_agent=self.user_agent or "PyGithub/Python", timeout=coalesce(self.timeout, github.MainClass.DEFAULT_TIMEOUT), # type: ignore[attr-defined] per_page=coalesce(self.per_page, github.MainClass.DEFAULT_PER_PAGE), # type: ignore[attr-defined] verify=coalesce(self.verify, True), retry=self.retry) @dataclasses.dataclass class GithubApp: """ Represents a GitHub application and all the required details. # Example ```py import os from github_bot_api import GithubApp with open(os.environ['PRIVATE_KEY_FILE']) as fp: private_key = fp.read() app = GithubApp( user_agent='my-bot/0.0.0', app_id=int(os.environ['APP_ID']), private_key=private_key) print(app.app_client().get_app().owner) ``` """ PUBLIC_GITHUB_V3_API_URL = 'https://api.github.com' #: User agent of the application. This will be respected in #get_user_agent(). user_agent: str #: GitHub Application ID. app_id: int #: RSA private key to sign the JWT with. private_key: str #: GitHub API base URL. Defaults to the public GitHub API. v3_api_url: str = PUBLIC_GITHUB_V3_API_URL def __post_init__(self): self._jwt_supplier = JwtSupplier(self.app_id, self.private_key) self._lock = threading.Lock() self._installation_tokens: t.Dict[int, InstallationTokenSupplier] = {} def _get_base_github_client_settings(self) -> GithubClientSettings: return GithubClientSettings(self.v3_api_url, self.get_user_agent()) def get_user_agent(self, installation_id: t.Optional[int] = None) -> str: """ Create a user agent string for the PyGithub client, including the installation if specified. """ user_agent = f'{self.user_agent} PyGithub/python (app_id={self.app_id}' if installation_id: user_agent += f', installation_id={installation_id})' return user_agent @property def jwt(self) -> TokenInfo: """ Returns the JWT for your GitHub application. The JWT is the token to use with GitHub application APIs. """ return self._jwt_supplier() @property def jwt_supplier(self) -> JwtSupplier: """ Returns a new #JwtSupplier that is used for generating JWT tokens for your GitHub application. """ return JwtSupplier(self.app_id, self.private_key) @property # type: ignore[misc] @deprecated.deprecated(reason='use GithubApp.app_client() instead', version='0.4.0') def client(self) -> 'github.Github': """ Use #app_client() instead. """ return self.app_client() def app_client(self, settings: t.Union[GithubClientSettings, t.Dict[str, t.Any], None] = None) -> 'github.Github': """ Returns a PyGithub client for your GitHub application. Note that the client's token will expire after 10 minutes and you will have to create a new client or update the client's token with the value returned by #jwt. It is recommended that you create a new client for each atomic operation you perform. This requires you to install `PyGithub`. """ if isinstance(settings, dict): settings = GithubClientSettings(**settings) elif settings is None: settings = GithubClientSettings() settings = self._get_base_github_client_settings().update(settings) return settings.make_client(jwt=self.jwt.value) def __requestor(self, auth_header: str, installation_id: int) -> t.Dict[str, str]: return requests.post( self.v3_api_url.rstrip('/') + f'/app/installations/{installation_id}/access_tokens', headers={'Authorization': auth_header, 'User-Agent': user_agent}, ).json() def get_installation_token_supplier(self, installation_id: int) -> InstallationTokenSupplier: """ Create an #InstallationTokenSupplier for your GitHub application to act within the scope of the given *installation_id*. """ with self._lock: return self._installation_tokens.setdefault( installation_id, InstallationTokenSupplier( self._jwt_supplier, installation_id, self.__requestor, ) ) def installation_token(self, installation_id: int) -> TokenInfo: """ A short-hand to retrieve a new installation token for the given *installation_id*. """ return self.get_installation_token_supplier(installation_id)() def installation_client( self, installation_id: int, settings: t.Union[GithubClientSettings, t.Dict[str, t.Any], None] = None, ) -> 'github.Github': """ Returns a PyGithub client for your GitHub application to act in the scope of the given *installation_id*. Note that the client's token will expire after 10 minutes and you will have to create a new client or update the client's token with the value returned by #jwt. It is recommended that you create a new client for each atomic operation you perform. This requires you to install `PyGithub`. """ if isinstance(settings, dict): settings = GithubClientSettings(**settings) elif settings is None: settings = GithubClientSettings() token = self.installation_token(installation_id).value settings = self._get_base_github_client_settings().update(settings) return settings.make_client(login_or_token=token)