Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Add an initial support for cron into CLI #128

Merged
merged 8 commits into from
Nov 29, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions .github/workflows/main.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ jobs:
- "3.10"
- "3.11"
- "3.12"
# - "3.13"
os:
- "ubuntu"
- "macos"
Expand Down
2 changes: 1 addition & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ skips = ["B102", "B701", "B507", "B601"]
[tool.vulture]
exclude = ["tests"]
ignore_decorators = []
ignore_names = []
ignore_names = ["verbose", "dry_run"]
make_whitelist = true
min_confidence = 80
paths = ["src/makim/"]
Expand Down
173 changes: 173 additions & 0 deletions src/makim/cli/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,173 @@
"""Cli functions to define the arguments and to 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,
create_dynamic_command_cron,
suggest_command,
)
from makim.cli.config import CLI_ROOT_FLAGS_VALUES_COUNT, extract_root_config
from makim.core import Makim

app = typer.Typer(
help=(
'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'
),
)

makim: Makim = Makim()


@app.callback(invoke_without_command=True)
def main(
ctx: typer.Context,
version: bool = typer.Option(
None,
'--version',
'-v',
is_flag=True,
help='Show the version and exit',
),
file: str = typer.Option(
'.makim.yaml',
'--file',
help='Makim config file',
),
dry_run: bool = typer.Option(
None,
'--dry-run',
is_flag=True,
help='Execute the command in dry mode',
),
verbose: bool = typer.Option(
None,
'--verbose',
is_flag=True,
help='Execute the command in verbose mode',
),
) -> None:
"""Process envers for specific flags, otherwise show the help menu."""
typer.echo(f'Makim file: {file}')

if version:
typer.echo(f'Version: {__version__}')
raise typer.Exit()

if ctx.invoked_subcommand is None:
typer.echo(ctx.get_help())
raise typer.Exit(0)


def _get_command_from_cli() -> str:
"""
Get the group and task from CLI.

This function is based on `CLI_ROOT_FLAGS_VALUES_COUNT`.
"""
params = sys.argv[1:]
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
break

idx += 1 + CLI_ROOT_FLAGS_VALUES_COUNT[arg]
except Exception as e:
print(e)

return command


def run_app() -> None:
"""Run the typer app."""
root_config = extract_root_config()

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
]

if not makim._check_makim_file(config_file_path) and cli_completion_words:
# autocomplete call
root_config = extract_root_config(cli_completion_words)
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)),
)

# 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

# Add dynamically cron commands to Typer app
if 'scheduler' in makim.global_data:
typer_cron = typer.Typer(
help='Tasks Scheduler',
invoke_without_command=True,
)

for schedule_name, schedule_params in makim.global_data.get(
'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')

# Add dynamically commands to 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
error_code = 2
if e.code != error_code:
raise e

command_used = _get_command_from_cli()

available_cmds = [
cmd.name for cmd in app.registered_commands if cmd.name is not None
]
suggestion = suggest_command(command_used, available_cmds)

typer.secho(
f"Command {command_used} not found. Did you mean '{suggestion}'?",
fg='red',
)

raise e


if __name__ == '__main__':
run_app()
Loading
Loading