Source code for _repobee.cli.dispatch

"""Module dispatching CLI commands to RepoBee's internal.

This module essentially translates parsed and processed arguments from the
CLI into commands for RepoBee's core.

.. module:: dispatch
    :synopsis: Command dispatcher for the CLI.

.. moduleauthor:: Simon Larsén
"""
import argparse
import pathlib
from typing import Optional, List, Mapping, NoReturn

import repobee_plug as plug

from _repobee import command, exception, formatters, util


[docs]def dispatch_command( args: argparse.Namespace, api: plug.PlatformAPI, config_file: pathlib.Path ) -> Mapping[str, List[plug.Result]]: """Handle parsed CLI arguments and dispatch commands to the appropriate functions. Expected exceptions are caught and turned into SystemExit exceptions, while unexpected exceptions are allowed to propagate. Args: args: A namespace of parsed command line arguments. api: An initialized plug.API instance. config_file: Path to the config file. """ hook_results: Mapping[str, List[plug.Result]] = {} dispatch_table = { plug.cli.CoreCommand.repos: _dispatch_repos_command, plug.cli.CoreCommand.issues: _dispatch_issues_command, plug.cli.CoreCommand.config: _dispatch_config_command, plug.cli.CoreCommand.reviews: _dispatch_reviews_command, plug.cli.CoreCommand.teams: _dispatch_teams_command, } is_ext_command = "_extension_command" in args if is_ext_command: ext_cmd = args._extension_command res = ( ext_cmd.command(api=api) if ext_cmd.__requires_api__() else ext_cmd.command() ) hook_results = ( {str(ext_cmd.__settings__.action): [res]} if res else hook_results ) else: category = args.category hook_results = ( dispatch_table[category](args, config_file, api) or hook_results ) if is_ext_command or args.action in [ plug.cli.CoreCommand.repos.setup, plug.cli.CoreCommand.repos.update, plug.cli.CoreCommand.repos.clone, ]: if hook_results: plug.echo(formatters.format_hook_results_output(hook_results)) if hook_results and "hook_results_file" in args and args.hook_results_file: _handle_hook_results( hook_results=hook_results, filepath=args.hook_results_file ) return hook_results
def _dispatch_repos_command( args: argparse.Namespace, config_file: pathlib.Path, api: plug.PlatformAPI ) -> Optional[Mapping[str, List[plug.Result]]]: repos = plug.cli.CoreCommand.repos action = args.action if action == repos.setup: return command.setup_student_repos( args.template_repo_urls, args.students, api ) elif action == repos.update: command.update_student_repos( args.template_repo_urls, args.students, api, issue=args.issue ) return None elif action == repos.migrate: command.migrate_repos(args.template_repo_urls, api) return None elif action == repos.clone: return command.clone_repos(args.repos, args.update_local, api) _raise_illegal_action_error(args) def _dispatch_issues_command( args: argparse.Namespace, config_file: pathlib.Path, api: plug.PlatformAPI ) -> Optional[Mapping[str, List[plug.Result]]]: issues = plug.cli.CoreCommand.issues action = args.action if action == issues.open: command.open_issue(args.issue, args.assignments, args.students, api) return None elif action == issues.close: command.close_issue(args.title_regex, args.repos, api) return None elif action == issues.list: return command.list_issues( args.repos, api, state=args.state, title_regex=args.title_regex or "", show_body=args.show_body, author=args.author, ) _raise_illegal_action_error(args) def _dispatch_config_command( args: argparse.Namespace, config_file: pathlib.Path, api: plug.PlatformAPI ) -> Optional[Mapping[str, List[plug.Result]]]: config = plug.cli.CoreCommand.config action = args.action if action == config.verify: plug.manager.hook.get_api_class().verify_settings( args.user, args.org_name, args.base_url, args.token, args.template_org_name, ) return None elif action == config.show: command.show_config(config_file, show_secrets=args.secrets) return None _raise_illegal_action_error(args) def _dispatch_reviews_command( args: argparse.Namespace, config_file: pathlib.Path, api: plug.PlatformAPI ) -> Optional[Mapping[str, List[plug.Result]]]: reviews = plug.cli.CoreCommand.reviews action = args.action if action == reviews.assign: command.assign_peer_reviews( args.assignments, args.students, args.num_reviews, args.issue, api ) return None elif action == reviews.end: command.purge_review_teams(args.assignments, args.students, api) return None elif action == reviews.check: command.check_peer_review_progress( args.assignments, args.students, args.title_regex, args.num_reviews, api, ) return None _raise_illegal_action_error(args) def _dispatch_teams_command( args: argparse.Namespace, config_file: pathlib.Path, api: plug.PlatformAPI ) -> Optional[Mapping[str, List[plug.Result]]]: teams = plug.cli.CoreCommand.teams action = args.action if action == teams.create: # list() is required here as the generator must be exhausted for teams # to be created list( plug.cli.io.progress_bar( command.create_teams( args.students, plug.TeamPermission.PUSH, api ), desc="Creating teams", unit="teams", total=len(args.students), ) ) return None _raise_illegal_action_error(args) def _raise_illegal_action_error(args: argparse.Namespace) -> NoReturn: raise exception.ParseError( f"Unknown action {args.action} for category {args.category}" ) def _handle_hook_results(hook_results, filepath): plug.log.warning( "Storing hook results to file is an alpha feature, the file format " "is not final" ) output_file = pathlib.Path(filepath) util.atomic_write(plug.result_mapping_to_json(hook_results), output_file) plug.echo("Hook results stored to {}".format(filepath))