Source code for octomachinery.app.action.runner

"""Octomachinery CLI runner for GitHub Action environments."""

import asyncio
import logging
from typing import Iterable, Optional

from aiohttp.client import ClientSession

# pylint: disable=relative-beyond-top-level
from ...github.entities.action import GitHubAction
# pylint: disable=relative-beyond-top-level
from ...github.errors import GitHubActionError
# pylint: disable=relative-beyond-top-level
from ...github.models.action_outcomes import (
    ActionFailure, ActionNeutral, ActionSuccess,
)
# pylint: disable=relative-beyond-top-level
from ..config import BotAppConfig
# pylint: disable=relative-beyond-top-level
from ..routing import WEBHOOK_EVENTS_ROUTER
# pylint: disable=relative-beyond-top-level
from ..routing.abc import OctomachineryRouterBase
# pylint: disable=relative-beyond-top-level
from ..routing.webhooks_dispatcher import route_github_event


logger = logging.getLogger(__name__)


[docs]async def process_github_action(config, event_routers): """Schedule GitHub Action event for processing.""" logger.info('Processing GitHub Action event...') async with ClientSession() as http_client_session: github_action = GitHubAction( metadata=config.action, http_session=http_client_session, config=config.github, event_routers=event_routers, ) logger.info('GitHub Action=%r', config.action) await route_github_event( github_event=github_action.event, github_app=github_action, ) return ActionSuccess('GitHub Action has been processed')
[docs]def run( *, config: Optional[BotAppConfig] = None, event_routers: Optional[Iterable[OctomachineryRouterBase]] = None, ) -> None: """Start up a server using CLI args for host and port.""" if event_routers is None: event_routers = {WEBHOOK_EVENTS_ROUTER} if config is None: config = BotAppConfig.from_dotenv() logging.basicConfig( level=logging.DEBUG if config.runtime.debug # pylint: disable=no-member else logging.INFO, ) try: processing_outcome = asyncio.run( process_github_action(config, event_routers), ) except GitHubActionError as action_error: action_error.terminate_action() except KeyboardInterrupt: ActionNeutral('Action processing interrupted by user').raise_it() except Exception: # pylint: disable=broad-except err_msg = 'Action processing failed unexpectedly' logger.exception(err_msg) ActionFailure(err_msg).raise_it() else: processing_outcome.raise_it()