Source code for octomachinery.github.api.app_client

"""GitHub App API client."""

from __future__ import annotations

import logging
from collections import defaultdict
from typing import TYPE_CHECKING, Any, Dict, Iterable

from aiohttp.client import ClientSession
from aiohttp.client_exceptions import ClientConnectorError

import attr
import sentry_sdk

# pylint: disable=relative-beyond-top-level
from ...routing import WEBHOOK_EVENTS_ROUTER
# pylint: disable=relative-beyond-top-level
from ...utils.asynctools import amap, dict_to_kwargs_cb
# pylint: disable=relative-beyond-top-level
from ..config.app import GitHubAppIntegrationConfig
# pylint: disable=relative-beyond-top-level
from ..entities.app_installation import GitHubAppInstallation
# pylint: disable=relative-beyond-top-level
from ..models import GitHubAppInstallation as GitHubAppInstallationModel
# pylint: disable=relative-beyond-top-level
from ..models.events import GitHubEvent
from .raw_client import RawGitHubAPI
from .tokens import GitHubJWTToken


if TYPE_CHECKING:
    # pylint: disable=relative-beyond-top-level
    from ...routing.abc import OctomachineryRouterBase


logger = logging.getLogger(__name__)


GH_INSTALL_EVENTS = {'integration_installation', 'installation'}


[docs]@attr.dataclass class GitHubApp: """GitHub API wrapper.""" _config: GitHubAppIntegrationConfig _http_session: ClientSession _event_routers: Iterable[OctomachineryRouterBase] = attr.ib( default={WEBHOOK_EVENTS_ROUTER}, converter=frozenset, ) def __attrs_post_init__(self) -> None: """Initialize the Sentry SDK library.""" # NOTE: Under the hood, it will set up the DSN from `SENTRY_DSN` # NOTE: env var. We don't need to care about it not existing as # NOTE: Sentry SDK helpers don't fail loudly and if not # NOTE: configured, it'll be ignored. # FIXME: # pylint: disable=fixme sentry_sdk.init() # pylint: disable=abstract-class-instantiated
[docs] async def dispatch_event(self, github_event: GitHubEvent) -> Iterable[Any]: """Dispatch ``github_event`` into the embedded routers.""" return await github_event.dispatch_via( *self._event_routers, # pylint: disable=not-an-iterable )
[docs] async def log_installs_list(self) -> None: """Store all installations data before starting.""" try: installations = await self.get_installations() except ClientConnectorError as client_error: logger.info('It looks like the GitHub API is offline...') logger.error( 'The following error has happened while trying to grab ' 'installations list: %s', client_error, ) return logger.info('This GitHub App is installed into:') # pylint: disable=protected-access for install_id, install_val in installations.items(): logger.info( '* Installation id %s (installed to %s)', install_id, install_val._metadata.account['login'], )
@property def gh_jwt(self): """Generate app's JSON Web Token, valid for 60 seconds.""" token = self._config.private_key.make_jwt_for( app_id=self._config.app_id, ) return GitHubJWTToken(token) @property def api_client(self): # noqa: D401 """The GitHub App client with an async CM interface.""" return RawGitHubAPI( token=self.gh_jwt, session=self._http_session, user_agent=self._config.user_agent, )
[docs] async def get_installation(self, event): """Retrieve an installation creds from store.""" if 'installation' not in event.payload: raise LookupError('This event occurred outside of an installation') install_id = event.payload['installation']['id'] return await self.get_installation_by_id(install_id)
[docs] async def get_installation_by_id(self, install_id): """Retrieve an installation with access tokens via API.""" return GitHubAppInstallation( GitHubAppInstallationModel( **( await self.api_client.getitem( '/app/installations/{installation_id}', url_vars={'installation_id': install_id}, preview_api_version='machine-man', ) ), ), self, )
[docs] async def get_installations(self): """Retrieve all installations with access tokens via API.""" installations: Dict[ int, GitHubAppInstallation, ] = defaultdict(dict) # type: ignore[arg-type] async for install in amap( dict_to_kwargs_cb(GitHubAppInstallationModel), self.api_client.getiter( '/app/installations', preview_api_version='machine-man', ), ): installations[install.id] = GitHubAppInstallation( install, self, ) return installations