From 5f43752555c92fd98c8aec73577b96cc565ebbcc Mon Sep 17 00:00:00 2001 From: Abhijeet Saroha Date: Sun, 22 Dec 2024 16:00:13 +0530 Subject: [PATCH] add support for cron commands --- src/makim/cli/__init__.py | 96 +++++++++++++++++++++------------------ 1 file changed, 52 insertions(+), 44 deletions(-) diff --git a/src/makim/cli/__init__.py b/src/makim/cli/__init__.py index 4860c90..0c0763f 100644 --- a/src/makim/cli/__init__.py +++ b/src/makim/cli/__init__.py @@ -1,14 +1,12 @@ -"""Cli functions to define the arguments and to call Makim.""" +"""CLI functions to define the arguments and call Makim.""" from __future__ import annotations import os import sys - from typing import Any, cast import typer - from makim import __version__ from makim.cli.auto_generator import ( create_dynamic_command, @@ -20,51 +18,50 @@ app = typer.Typer( help=( - 'Makim is a tool that helps you to organize ' - 'and simplify your helper commands.' + "Makim is a tool that helps you to organize " + "and simplify your helper commands." ), epilog=( - 'If you have any problem, open an issue at: ' - 'https://github.com/osl-incubator/makim' + "If you have any problem, open an issue at: " + "https://github.com/osl-incubator/makim" ), ) makim: Makim = Makim() - @app.callback(invoke_without_command=True) def main( ctx: typer.Context, version: bool = typer.Option( None, - '--version', - '-v', + "--version", + "-v", is_flag=True, - help='Show the version and exit', + help="Show the version and exit", ), file: str = typer.Option( - '.makim.yaml', - '--file', - help='Makim config file', + ".makim.yaml", + "--file", + help="Makim config file", ), dry_run: bool = typer.Option( None, - '--dry-run', + "--dry-run", is_flag=True, - help='Execute the command in dry mode', + help="Execute the command in dry mode", ), verbose: bool = typer.Option( None, - '--verbose', + "--verbose", is_flag=True, - help='Execute the command in verbose mode', + help="Execute the command in verbose mode", ), ) -> None: - """Process envers for specific flags, otherwise show the help menu.""" - typer.echo(f'Makim file: {file}') + """Process top-level flags; otherwise, show the help menu.""" + typer.echo(f"Makim file: {file}") if version: - typer.echo(f'Version: {__version__}') + typer.echo(f"Version: {__version__}") raise typer.Exit() if ctx.invoked_subcommand is None: @@ -79,14 +76,14 @@ def _get_command_from_cli() -> str: This function is based on `CLI_ROOT_FLAGS_VALUES_COUNT`. """ params = sys.argv[1:] - command = '' + command = "" try: idx = 0 while idx < len(params): arg = params[idx] if arg not in CLI_ROOT_FLAGS_VALUES_COUNT: - command = f'flag `{arg}`' if arg.startswith('--') else arg + command = f"flag `{arg}`" if arg.startswith("--") else arg break idx += 1 + CLI_ROOT_FLAGS_VALUES_COUNT[arg] @@ -97,59 +94,70 @@ def _get_command_from_cli() -> str: def run_app() -> None: - """Run the typer app.""" + """Run the Typer app.""" root_config = extract_root_config() - config_file_path = cast(str, root_config.get('file', '.makim.yaml')) + config_file_path = cast(str, root_config.get("file", ".makim.yaml")) cli_completion_words = [ - w for w in os.getenv('COMP_WORDS', '').split('\n') if w + w for w in os.getenv("COMP_WORDS", "").split("\n") if w ] if not makim._check_makim_file(config_file_path) and cli_completion_words: - # autocomplete call + # Autocomplete call root_config = extract_root_config(cli_completion_words) - config_file_path = cast(str, root_config.get('file', '.makim.yaml')) + config_file_path = cast(str, root_config.get("file", ".makim.yaml")) if not makim._check_makim_file(config_file_path): return makim.load( file=config_file_path, - dry_run=cast(bool, root_config.get('dry_run', False)), - verbose=cast(bool, root_config.get('verbose', False)), + dry_run=cast(bool, root_config.get("dry_run", False)), + verbose=cast(bool, root_config.get("verbose", False)), ) - # create tasks data + # Create tasks data tasks: dict[str, Any] = {} - for group_name, group_data in makim.global_data.get('groups', {}).items(): - for task_name, task_data in group_data.get('tasks', {}).items(): - tasks[f'{group_name}.{task_name}'] = task_data + for group_name, group_data in makim.global_data.get("groups", {}).items(): + for task_name, task_data in group_data.get("tasks", {}).items(): + tasks[f"{group_name}.{task_name}"] = task_data - # Add dynamically cron commands to Typer app - if 'scheduler' in makim.global_data: + # Add dynamically created cron commands + if "scheduler" in makim.global_data: typer_cron = typer.Typer( - help='Tasks Scheduler', + help="Tasks Scheduler", invoke_without_command=True, ) for schedule_name, schedule_params in makim.global_data.get( - 'scheduler', {} + "scheduler", {} ).items(): create_dynamic_command_cron( makim, typer_cron, schedule_name, schedule_params or {} ) - # Add cron command - app.add_typer(typer_cron, name='cron', rich_help_panel='Extensions') + @typer_cron.command(help="List all scheduled tasks") + def list(): + print("list") + + @typer_cron.command(help="Start a scheduler by its name") + def start(name: str): + print("start") + + @typer_cron.command(help="Stop a scheduler by its name") + def stop(name: str): + print("Stop") + + app.add_typer(typer_cron, name="cron", rich_help_panel="Extensions") - # Add dynamically commands to Typer app + # Add dynamically created commands to the Typer app for name, args in tasks.items(): create_dynamic_command(makim, app, name, args) try: app() except SystemExit as e: - # code 2 means code not found + # Code 2 means command not found error_code = 2 if e.code != error_code: raise e @@ -163,11 +171,11 @@ def run_app() -> None: typer.secho( f"Command {command_used} not found. Did you mean '{suggestion}'?", - fg='red', + fg="red", ) raise e -if __name__ == '__main__': +if __name__ == "__main__": run_app()