Skip to content

Commit

Permalink
add support for cron commands
Browse files Browse the repository at this point in the history
  • Loading branch information
abhijeetSaroha committed Dec 22, 2024
1 parent 8450256 commit 5f43752
Showing 1 changed file with 52 additions and 44 deletions.
96 changes: 52 additions & 44 deletions src/makim/cli/__init__.py
Original file line number Diff line number Diff line change
@@ -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,
Expand All @@ -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:
Expand All @@ -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]
Expand All @@ -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
Expand All @@ -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()

0 comments on commit 5f43752

Please sign in to comment.