Task Manager CLI

Command line tools for TaskManager applications.

This module requires the cli extra to be installed:

pip install aio-fluid[cli]

Setup

Wrap your FastAPI app with TaskManagerCLI and call it as the entry point:

from fluid.scheduler.cli import TaskManagerCLI

task_manager_cli = TaskManagerCLI("examples.tasks:task_app")

if __name__ == "__main__":
    task_manager_cli()

task_manager_app can be a FastAPI instance, a callable that returns one, or a dotted-import string (resolved at call time so the CLI stays importable without loading the full app).

Commands

ls — list registered tasks

Prints a table of all registered tasks with their schedule, priority, timeout, CPU-bound flag, and tags.

python -m myapp ls

Pass --tags/-t (repeatable) to only list tasks that have at least one of the given tags:

python -m myapp ls --tags fast --tags slow

serve — start the HTTP server

python -m myapp serve --host 0.0.0.0 --port 8080

Delegates to uvicorn. Pass --reload for development auto-reload.

enable — enable or disable a task

python -m myapp enable <task-name>           # enable
python -m myapp enable <task-name> --disable  # disable

exec — execute a task

Runs a registered task synchronously and prints the result table.

python -m myapp exec <task-name> [OPTIONS]

Each top level field of the task params model is a CLI option, named after the field with underscores turned into dashes:

python -m myapp exec add --a 5 --b 3

Values are written as they are, not as JSON, and validated by the params model:

• strings, emails, decimals, dates, datetimes and secrets are plain strings, --email luca@example.com, --dt 2026-01-01T10:00:00
• integers and floats are numbers, --count 3
• booleans are flags, --error or --no-error
• enums and literals accept one of their values, --color green
• a list of plain values is passed by repeating the option, --tag a --tag b
• structured values, such as a nested model or a dictionary, are passed as JSON

Options left out are not sent, so the model defaults apply. A required field can be given either as an option or in --params.

Passing parameters as JSON

All task parameters can also be supplied together as a JSON string with --params. Values in --params always take priority over individual CLI option defaults:

python -m myapp exec add --params '{"a": 5, "b": 3}'

This is useful when parameters are complex types or when scripting task execution. If both --params and individual options are provided, the JSON values win for any overlapping keys.

# error=true from --params overrides the model default error=false
python -m myapp exec fast --params '{"error": true}'

--run-id

Pass a custom run ID to correlate the task run with external systems:

python -m myapp exec add --run-id my-custom-id --params '{"a": 1, "b": 2}'

--log

Enable structured logging output during execution:

python -m myapp exec add --log --params '{"a": 1, "b": 2}'

--params-from-broker

Load the parameters the task consumer stored in the broker for --run-id. This is how a CPU bound task receives its parameters: the consumer stores them before starting the process, or the Kubernetes Job, and the process reads and removes them. They are kept off the command line because any user on the host, or anyone who can read the Job, can see it. Parameters given as options or with --params are applied on top.

API reference

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

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 {}

get_task_manager_app

get_task_manager_app()

Get the FastAPI app for the TaskManager.

Source code in fluid/scheduler/cli.py

def get_task_manager_app(self) -> FastAPI:
    """Get the FastAPI app for the TaskManager."""
    if isinstance(self.task_manager_app, str):
        return import_from_string(self.task_manager_app)()
    elif isinstance(self.task_manager_app, FastAPI):
        return self.task_manager_app
    else:
        return self.task_manager_app()

get_task_manager

get_task_manager()

Get the TaskManager instance from the app state.

Source code in fluid/scheduler/cli.py

def get_task_manager(self) -> TaskManager:
    """Get the TaskManager instance from the app state."""
    app = self.get_task_manager_app()
    if not hasattr(app.state, "task_manager"):
        raise RuntimeError("The app does not have a task_manager in its state")
    return app.state.task_manager

list_commands

list_commands(ctx)

Source code in fluid/utils/lazy.py

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

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)