Skip to content

Task Manager Cli

Command line tools for task manager applications.

This modules requires the cli extra to be installed.

$ pip install aio-fluid[cli]
It can be imported from fluid.scheduler.cli:

from fastapi.scheduler.cli import TaskManagerCLI

if __name__ == "__main__":
    cli = TaskManagerCLI("path.to:task_app")
    cli()

fluid.scheduler.cli.TaskManagerCLI 

TaskManagerCLI(task_manager_app, **kwargs)

Bases: LazyGroup

CLI for TaskManager

This class provides a CLI for a TaskManager Application.

It requires to install the cli extra dependencies.

Source code in fluid/scheduler/cli.py 
37
38
39
40
41
42
43
44
def __init__(
    self,
    task_manager_app: TaskManagerApp,
    **kwargs: Any,
):
    kwargs.setdefault("commands", DEFAULT_COMMANDS)
    super().__init__(**kwargs)
    self.task_manager_app = task_manager_app

lazy_subcommands instance-attribute 

lazy_subcommands = lazy_subcommands or {}

task_manager_app instance-attribute 

task_manager_app = task_manager_app

list_commands 

list_commands(ctx)
Source code in fluid/utils/lazy.py 
28
29
30
31
def list_commands(self, ctx: click.Context) -> list[str]:
    commands = super().list_commands(ctx)
    commands.extend(self.lazy_subcommands)
    return sorted(commands)

get_command 

get_command(ctx, cmd_name)
Source code in fluid/utils/lazy.py 
33
34
35
36
def get_command(self, ctx: click.Context, cmd_name: str) -> click.Command | None:
    if cmd_name in self.lazy_subcommands:
        return self._lazy_load(cmd_name)
    return super().get_command(ctx, cmd_name)