plexus.cli.shared.CommandDispatch module

class plexus.cli.shared.CommandDispatch.ItemCountColumn(table_column: Column | None = None)

Bases: ProgressColumn

Renders item count and total.

render(task: Task) → str

Show item count and total.

class plexus.cli.shared.CommandDispatch.OrderCommands(name: str | None = None, commands: MutableMapping[str, Command] | Sequence[Command] | None = None, invoke_without_command: bool = False, no_args_is_help: bool | None = None, subcommand_metavar: str | None = None, chain: bool = False, result_callback: Callable[[...], Any] | None = None, **kwargs: Any)

Bases: Group

list_commands(ctx: Context) → list[str]

Returns a list of subcommand names in the order they should appear.

class plexus.cli.shared.CommandDispatch.StatusColumn(table_column: Column | None = None)

Bases: ProgressColumn

Renders status in a full line above the progress bar.

render(task: Task) → str

Should return a renderable object.

plexus.cli.shared.CommandDispatch.create_celery_app() → Celery

Create a configured Celery application with AWS credentials.

plexus.cli.shared.CommandDispatch.create_cli()

Create and configure the command line interface.

plexus.cli.shared.CommandDispatch.ensure_tasks_registered()

Ensure tasks are registered, registering them lazily on first access.

plexus.cli.shared.CommandDispatch.get_celery_app()

Get the Celery app instance, creating it lazily on first access.

plexus.cli.shared.CommandDispatch.get_progress_status(current: int, total: int) → str

Get a status message based on progress percentage.

plexus.cli.shared.CommandDispatch.safequote(value: str) → str

Safely quote a string value, handling None.