Skip to content

Task Manager Cli

Command line tools for TaskManager 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, log_config=None, **kwargs)

Bases: LazyGroup

CLI for TaskManager

This class provides a CLI for a TaskManager Application.

It requires to install the cli extra dependencies.

PARAMETER DESCRIPTION
task_manager_app

Task manager application.

This can be a FastAPI app, a callable that returns a FastAPI app, or a string import path to a FastAPI app.

TYPE: TaskManagerApp

log_config

Log configuration parameters.

These parameters are passed to the log.config function when configuring logging.

TYPE: dict | None DEFAULT: None

Source code in fluid/scheduler/cli.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    task_manager_app: Annotated[
        TaskManagerApp,
        Doc(
            """
            Task manager application.

            This can be a FastAPI app, a callable that returns a FastAPI app,
            or a string import path to a FastAPI app.
            """
        ),
    ],
    log_config: Annotated[
        dict | None,
        Doc(
            """
            Log configuration parameters.

            These parameters are passed to the [log.config][fluid.utils.log.config]
            function when configuring logging.
            """
        ),
    ] = None,
    **kwargs: Any,
):
    kwargs.setdefault("commands", DEFAULT_COMMANDS)
    super().__init__(**kwargs)
    self.task_manager_app = task_manager_app
    self.log_config = log_config or {}

task_manager_app instance-attribute 

task_manager_app = task_manager_app

log_config instance-attribute 

log_config = log_config or {}

lazy_subcommands instance-attribute 

lazy_subcommands = lazy_subcommands or {}

list_commands 

list_commands(ctx)
Source code in fluid/utils/lazy.py 
45
46
47
48
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 
50
51
52
53
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)