process.deptree

process.deptree

Classes

Name	Description
Task	Base class for tasks implementing a dependency tree structure for execution workflow.

Task

process.deptree.Task()

Base class for tasks implementing a dependency tree structure for execution workflow.

This class provides the foundation for creating tasks that can be organized in a dependency graph and executed in the correct order. Tasks can be generated and run using the gen function with different execution modes (sync, executor, or prefect).

Attributes: output: Optional attribute for storing the task’s output path or result. deps: Optional list of Task instances that this task depends on. status: Task execution status

Methods: dependencies(): Returns the list of dependent tasks. The default implementation returns the deps attribute, if it exists. run(): Implements the actual execution code done(): Checks if the task is completed. The default implementation checks if the output Path exists, if defined. cleanup(): Optional cleanup method executed when all parent tasks have finished

Usage: Subclass Task and implement at minimum: - The run() method with your task’s execution logic (if any) - Optionally override dependencies() or self.deps if your task has dependencies - Optionally override cleanup() if your task needs cleanup after execution - Set output attribute if your task produces output files

Example: See the sample function for a complete example of Task definition and usage.

Methods

Name	Description
cleanup	Optional cleanup method executed after the task finishes running
dependencies	Returns a list of dependencies, each being a Task.
done	Whether the task is done
run	Implements the actual execution code

cleanup

process.deptree.Task.cleanup()

Optional cleanup method executed after the task finishes running

dependencies

process.deptree.Task.dependencies()

Returns a list of dependencies, each being a Task.

done

process.deptree.Task.done()

Whether the task is done

run

process.deptree.Task.run()

Implements the actual execution code

Functions

Name	Description
gen	Run a task and its dependencies
gen_executor	Recursively generate dependencies of task, then call its run method in a
gen_sync	Run a Task in sequential mode
print_execution_summary	Print a summary table showing count of tasks by status and class name.
run_with_status_update	Wrapper function that sets status to running when actually executing
sample	Generate a sample Product for demonstration purposes

gen

process.deptree.gen(task, mode='sync', **kwargs)

Run a task and its dependencies

Arguments: mode (str): - ‘sync’: sequential - ‘executor’: using executors - ‘prefect’: using prefect as a worflow manager

gen_executor

process.deptree.gen_executor(
    task,
    executors=None,
    default_executor=None,
    graph_executor=None,
    verbose=False,
    _tasks=None,
    log_file=None,
)

Recursively generate dependencies of task, then call its run method in a custom executor.

Args: task: a Task object executors: dictionary of executors default_executor: the executor to use for all tasks not included in executors. Defaults to ThreadPoolExecutor graph_executor: an instance of ThreadPoolExecutor used for traversing the dependency tree with the gen function. It is useful to override the default ThreadPoolExecutor(512) if the tree is too large. _tasks: internal parameter to track tasks status and provide an execution summary. Also used to avoid duplicate task submissions.

Example: from concurrent.futures import ThreadPoolExecutor from distributed import Client # dask executor

gen(task,
    executors={'Level1' : ThreadPoolExecutor(2)},
    default_executor=Client()  # dask executor
        # Note: can also use dask_jobqueue to instantiate
        # an executor backed by a job queuing system like
        # HTCondor
})

For details on how to implement task, please check function core.deptree.sample.

gen_sync

process.deptree.gen_sync(task, verbose=False)

Run a Task in sequential mode

Args: task (Task): The task to generate verbose (bool): Whether to print verbose output

print_execution_summary

process.deptree.print_execution_summary(
    tasks,
    start_time,
    log_file=None,
    finished=False,
)

Print a summary table showing count of tasks by status and class name.

Args: tasks: List of Task instances to summarize

run_with_status_update

process.deptree.run_with_status_update(task, verbose)

Wrapper function that sets status to running when actually executing

sample

process.deptree.sample()

Generate a sample Product for demonstration purposes