ddp_utils.console.task_manager¶

import ddp_utils.console.task_manager

Build hierarchical tasks on the single coordinated console renderer.

No Rich Live instance or secondary cursor owner is created here. Progress, spinner, grouping, logging, and cleanup all use ConsoleLayout.

Examples

Run nested progress with one render owner:

with LiveProgressManager() as manager:
    job = manager.add_main_task_obj("job", "Cases", 2)
    with job.create_spinner("login", "Signing in"):
        authenticate()
    job.update()

class ddp_utils.console.task_manager.BaseTask(manager: LiveProgressManager, task_id: str)¶

Bases: object

Reference one manager-owned task by identifier.

Parameters

Name	Type	Description
manager	LiveProgressManager	Owning task manager.
task_id	str	Registered task identifier.

Examples

Remove a task through its handle:

task.remove()

Initialize a handle for one manager-owned task.

Parameters

Name	Type	Description
manager	LiveProgressManager	Task manager that owns the referenced task.
task_id	str	Unique identifier registered in `manager`.

Examples

Create a handle for an existing task:

task = BaseTask(manager, "download")

remove() → None¶

Remove the task and its descendants atomically.

Examples

Clear a finished task block:

task.remove()

class ddp_utils.console.task_manager.LiveProgressManager(debug: bool = False, expand_tables: bool = True, use_panels: bool = False, panel_box: str = 'SIMPLE', indent_step: int = 2, main_config: Dict | None = None, autostart: bool = True, console: Console | None = None)¶

Bases: object

Manage nested progress and spinners through one console owner.

The historical name is retained, but this class never constructs Rich Live. Manual and context-managed usage share the same lifecycle.

Parameters

Name	Type	Description
debug	bool	Retained compatibility flag.
expand_tables	bool	Retained compatibility flag.
use_panels	bool	Retained compatibility flag.
panel_box	str	Retained compatibility setting.
indent_step	int	Spaces added per nesting level.
main_config	Optional[Dict]	Default top-level progress options.
autostart	bool	Start the root group immediately.
console	Optional[Console]	Optional owning console.

Examples

Manage lifecycle explicitly:

manager = LiveProgressManager(autostart=False)
manager.start()
manager.stop()

Initialize the shared hierarchical task manager.

Parameters

Name	Type	Description
debug	bool	Retained compatibility flag exposed to existing callers.
expand_tables	bool	Retained table-expansion compatibility setting.
use_panels	bool	Retained panel-rendering compatibility setting.
panel_box	str	Retained panel border style name.
indent_step	int	Number of spaces added for each nesting level.
main_config	Optional[Dict]	Default options merged into top-level progress tasks.
autostart	bool	Start the root console group during initialization.
console	Optional[Console]	Console owner, or `None` to use the process singleton.

Examples

Create a manager whose lifecycle starts explicitly:

manager = LiveProgressManager(autostart=False)
manager.start()

start() → LiveProgressManager¶

Start the shared root group once.

Returns

Type	Description
LiveProgressManager	This manager.

Examples

Start explicit lifecycle management:

manager.start()

stop() → None¶

Remove all tasks and the root group.

Examples

Release manager output:

manager.stop()

create_progress(task_id: str, label: str, total: int = 1, parent: str | None = None, config: Dict | None = None) → ProgressTask¶

Create a root or child progress task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
label	str	Visible label.
total	int	Initial total.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
ProgressTask	Active task handle.

Examples

Create nested row progress:

task = manager.create_progress("rows", "Rows", 20, "job")

add_progress(task_id: str, label: str, total: int = 1, parent: str | None = None, config: Dict | None = None) → ProgressTask¶

Create a root or child progress task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
label	str	Visible label.
total	int	Initial total.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
ProgressTask	Active task handle.

Examples

Create nested row progress:

task = manager.create_progress("rows", "Rows", 20, "job")

add_main_task(task_id: str, label: str, total: int) → None¶

Create a top-level task without returning its handle.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
label	str	Visible label.
total	int	Initial total.

Examples

Add overall progress:

manager.add_main_task("job", "Cases", 10)

add_main_task_obj(task_id: str, label: str, total: int) → MainTask¶

Create and return a top-level task handle.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
label	str	Visible label.
total	int	Initial total.

Returns

Type	Description
MainTask	Main task handle.

Examples

Create object-style progress:

task = manager.add_main_task_obj("job", "Cases", 10)

create_spinner(task_id: str, message: str, parent: str | None = None, config: Dict | None = None) → SpinnerTask¶

Create a root or child spinner task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
message	str	Visible message.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
SpinnerTask	Active spinner handle.

Examples

Create nested connection status:

spinner = manager.create_spinner("http", "Connecting", "job")

add_spinner(task_id: str, message: str, parent: str | None = None, config: Dict | None = None) → SpinnerTask¶

Create a root or child spinner task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
message	str	Visible message.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
SpinnerTask	Active spinner handle.

Examples

Create nested connection status:

spinner = manager.create_spinner("http", "Connecting", "job")

update(task_id: str, advance: int = 1) → None¶

Advance a progress task.

Parameters

Name	Type	Description
task_id	str	Progress identifier.
advance	int	Units to add.

Examples

Advance five rows:

manager.update("rows", 5)

progress_next(task_id: str, advance: int = 1) → None¶

Advance a progress task.

Parameters

Name	Type	Description
task_id	str	Progress identifier.
advance	int	Units to add.

Examples

Advance five rows:

manager.update("rows", 5)

update_main(task_id: str, advance: int = 1) → None¶

Advance a progress task.

Parameters

Name	Type	Description
task_id	str	Progress identifier.
advance	int	Units to add.

Examples

Advance five rows:

manager.update("rows", 5)

update_total(task_id: str, total: int) → None¶

Replace a progress total.

Parameters

Name	Type	Description
task_id	str	Progress identifier.
total	int	New non-negative total.

Examples

Apply a discovered total:

manager.update_total("rows", 500)

set_main_total(task_id: str, total: int) → None¶

Replace a progress total.

Parameters

Name	Type	Description
task_id	str	Progress identifier.
total	int	New non-negative total.

Examples

Apply a discovered total:

manager.update_total("rows", 500)

complete(task_id: str) → None¶

Complete and remove a progress task.

Parameters

Name	Type	Description
task_id	str	Progress identifier.

Examples

Complete successful work:

manager.complete("rows")

stop_spinner(task_id: str) → None¶

Stop and remove a spinner task.

Parameters

Name	Type	Description
task_id	str	Spinner identifier.

Examples

Stop connection status:

manager.stop_spinner("http")

remove(task_id: str) → None¶

Remove a task and descendants; ignore unknown identifiers.

Parameters

Name	Type	Description
task_id	str	Task identifier.

Examples

Clear nested output:

manager.remove("job")

remove_main(task_id: str) → None¶

Remove a task and descendants; ignore unknown identifiers.

Parameters

Name	Type	Description
task_id	str	Task identifier.

Examples

Clear nested output:

manager.remove("job")

progress_bar(task_id: str, label: str, total: int = 1, parent: str | None = None, config: Dict | None = None) → Iterator[LiveProgressManager]¶

Manage one temporary progress task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
label	str	Visible label.
total	int	Initial total.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Yields

This manager for update calls.

Raises

Exception	Description
Exception	Re-raises any exception from the managed block after removing the temporary progress task.

Examples

Advance temporary progress:

with manager.progress_bar("rows", "Rows", 10) as progress:
    progress.progress_next("rows")

spinner(task_id: str, message: str, parent: str | None = None, config: Dict | None = None) → Iterator[LiveProgressManager]¶

Manage one temporary spinner task.

Parameters

Name	Type	Description
task_id	str	Unique identifier.
message	str	Visible message.
parent	str \| None	Optional parent identifier.
config	Dict \| None	Optional rendering overrides.

Yields

This manager while status is active.

Examples

Display startup status:

with manager.spinner("browser", "Starting"):
    launch_browser()

cleanup_all() → None¶

Remove all current tasks while keeping the manager reusable.

Examples

Clear one completed batch:

manager.cleanup_all()

class ddp_utils.console.task_manager.MainTask(manager: LiveProgressManager, task_id: str)¶

Bases: ProgressTask

Represent a top-level progress task.

Examples

Obtain a main task from the manager:

task = manager.add_main_task_obj("job", "Cases", 100)

Initialize a handle for one manager-owned task.

Parameters

Name	Type	Description
manager	LiveProgressManager	Task manager that owns the referenced task.
task_id	str	Unique identifier registered in `manager`.

Examples

Create a handle for an existing task:

task = BaseTask(manager, "download")

ddp_utils.console.task_manager.ProgressManager¶: alias of LiveProgressManager

class ddp_utils.console.task_manager.ProgressTask(manager: LiveProgressManager, task_id: str)¶

Bases: BaseTask

Control progress and create nested child tasks.

Examples

Advance returned progress:

task.update(5)

Initialize a handle for one manager-owned task.

Parameters

Name	Type	Description
manager	LiveProgressManager	Task manager that owns the referenced task.
task_id	str	Unique identifier registered in `manager`.

Examples

Create a handle for an existing task:

task = BaseTask(manager, "download")

update(advance: int = 1) → None¶

Advance progress by a number of units.

Parameters

Name	Type	Description
advance	int	Units to add.

Examples

Record two completed items:

task.update(2)

set_total(total: int) → None¶

Replace the task total.

Parameters

Name	Type	Description
total	int	New non-negative total.

Examples

Apply a discovered total:

task.set_total(250)

complete() → None¶

Complete progress and remove the task group.

Examples

Finalize successful work:

task.complete()

create_progress(task_id: str, label: str, total: int = 1, config: Dict | None = None) → ProgressTask¶

Create a child progress task.

Parameters

Name	Type	Description
task_id	str	Unique child identifier.
label	str	Visible label.
total	int	Initial total.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
ProgressTask	Active child handle.

Examples

Create nested page progress:

child = task.create_progress("pages", "Pages", 10)

add_progress(task_id: str, label: str, total: int = 1, config: Dict | None = None) → ProgressTask¶

Create a child progress task.

Parameters

Name	Type	Description
task_id	str	Unique child identifier.
label	str	Visible label.
total	int	Initial total.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
ProgressTask	Active child handle.

Examples

Create nested page progress:

child = task.create_progress("pages", "Pages", 10)

create_spinner(task_id: str, message: str, config: Dict | None = None) → SpinnerTask¶

Create a child spinner.

Parameters

Name	Type	Description
task_id	str	Unique child identifier.
message	str	Visible status message.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
SpinnerTask	Active spinner handle.

Examples

Create nested authentication status:

spinner = task.create_spinner("auth", "Signing in")

add_spinner(task_id: str, message: str, config: Dict | None = None) → SpinnerTask¶

Create a child spinner.

Parameters

Name	Type	Description
task_id	str	Unique child identifier.
message	str	Visible status message.
config	Dict \| None	Optional rendering overrides.

Returns

Type	Description
SpinnerTask	Active spinner handle.

Examples

Create nested authentication status:

spinner = task.create_spinner("auth", "Signing in")

class ddp_utils.console.task_manager.SpinnerTask(manager: LiveProgressManager, task_id: str)¶

Bases: BaseTask

Control one spinner rendered by the unified console.

Examples

Update a spinner message:

spinner.update_message("Downloading")

Initialize a handle for one manager-owned task.

Parameters

Name	Type	Description
manager	LiveProgressManager	Task manager that owns the referenced task.
task_id	str	Unique identifier registered in `manager`.

Examples

Create a handle for an existing task:

task = BaseTask(manager, "download")

stop() → None¶

Stop and remove this spinner.

Examples

Clear completed spinner output:

spinner.stop()

complete() → None¶

Stop and remove this spinner.

Examples

Clear completed spinner output:

spinner.stop()

update_message(message: str) → None¶

Replace the current message.

Parameters

Name	Type	Description
message	str	New status text.

Examples

Publish a new phase:

spinner.update_message("Waiting")