luigi.worker

The worker communicates with the scheduler and does two things:

1. Sends all tasks that has to be run

2. Gets tasks from the scheduler that should be run

When running in local mode, the worker talks directly to a Scheduler instance. When you run a central server, the worker will talk to the scheduler using a RemoteScheduler instance.

Everything in this module is private to luigi and may change in incompatible ways between versions. The exception is the exception types and the worker config class.

Functions

check_complete(task, out_queue[, ...])	Checks if task is complete, puts the result to out_queue, optionally using the completion cache.
check_complete_cached(task[, completion_cache])
rpc_message_callback(fn)

Classes

ContextManagedTaskProcess(context, *args, ...)
DequeQueue	deque wrapper implementing the Queue interface.
GetWorkResponse(task_id, running_tasks, ...)	Create new instance of GetWorkResponse(task_id, running_tasks, n_pending_tasks, n_unique_pending, n_pending_last_scheduled, worker_state)
KeepAliveThread(scheduler, worker_id, ...)	Periodically tell the scheduler that the worker still lives.
SchedulerMessage(scheduler, task_id, ...)	Message object that is build by the the Worker when a message from the scheduler is received and passed to the message queue of a Task.
SingleProcessPool()	Dummy process pool for using a single processor.
TaskProcess(task, worker_id, result_queue, ...)	Wrap all task execution in this class.
TaskStatusReporter(scheduler, task_id, ...)	Reports task status information to the scheduler.
TracebackWrapper(trace)	Class to wrap tracebacks so we can know they're not just strings.
Worker([scheduler, worker_id, ...])	Worker object communicates with a scheduler.
worker(*args, **kwargs)

Exceptions

AsyncCompletionException(trace)	Exception indicating that something went wrong with checking complete.
TaskException

exception luigi.worker.TaskException

class luigi.worker.GetWorkResponse(task_id, running_tasks, n_pending_tasks, n_unique_pending, n_pending_last_scheduled, worker_state)

Create new instance of GetWorkResponse(task_id, running_tasks, n_pending_tasks, n_unique_pending, n_pending_last_scheduled, worker_state)

n_pending_last_scheduled

Alias for field number 4

n_pending_tasks

Alias for field number 2

n_unique_pending

Alias for field number 3

running_tasks

Alias for field number 1

task_id

Alias for field number 0

worker_state

Alias for field number 5

class luigi.worker.TaskProcess(task, worker_id, result_queue, status_reporter, use_multiprocessing=False, worker_timeout=0, check_unfulfilled_deps=True, check_complete_on_run=False, task_completion_cache=None)

Wrap all task execution in this class.

Mainly for convenience since this is run in a separate process.

forward_reporter_attributes = {'decrease_running_resources': 'decrease_running_resources', 'scheduler_messages': 'scheduler_messages', 'update_progress_percentage': 'set_progress_percentage', 'update_status_message': 'set_status_message', 'update_tracking_url': 'set_tracking_url'}

run()

Method to be run in sub-process; can be overridden in sub-class

terminate()

Terminate this process and its subprocesses.

class luigi.worker.ContextManagedTaskProcess(context, *args, **kwargs)

run()

Method to be run in sub-process; can be overridden in sub-class

class luigi.worker.TaskStatusReporter(scheduler, task_id, worker_id, scheduler_messages)

Reports task status information to the scheduler.

This object must be pickle-able for passing to TaskProcess on systems where fork method needs to pickle the process object (e.g. Windows).

update_tracking_url(tracking_url)

update_status_message(message)

update_progress_percentage(percentage)

decrease_running_resources(decrease_resources)

report_task_statistics(statistics)

class luigi.worker.SchedulerMessage(scheduler, task_id, message_id, content, **payload)

Message object that is build by the the Worker when a message from the scheduler is received and passed to the message queue of a Task.

respond(response)

class luigi.worker.SingleProcessPool

Dummy process pool for using a single processor.

Imitates the api of multiprocessing.Pool using single-processor equivalents.

apply_async(function, args)

close()

join()

class luigi.worker.DequeQueue

deque wrapper implementing the Queue interface.

put(obj, block=None, timeout=None)

get(block=None, timeout=None)

exception luigi.worker.AsyncCompletionException(trace)

Exception indicating that something went wrong with checking complete.

class luigi.worker.TracebackWrapper(trace)

Class to wrap tracebacks so we can know they’re not just strings.

luigi.worker.check_complete_cached(task, completion_cache=None)

luigi.worker.check_complete(task, out_queue, completion_cache=None)

Checks if task is complete, puts the result to out_queue, optionally using the completion cache.

class luigi.worker.worker(*args, **kwargs)

id

Parameter whose value is a str, and a base class for other parameter types.

Parameters are objects set on the Task class level to make it possible to parameterize tasks. For instance:

class MyTask(luigi.Task):
    foo = luigi.Parameter()

class RequiringTask(luigi.Task):
    def requires(self):
        return MyTask(foo="hello")

    def run(self):
        print(self.requires().foo)  # prints "hello"

This makes it possible to instantiate multiple tasks, eg MyTask(foo='bar') and MyTask(foo='baz'). The task will then have the foo attribute set appropriately.

When a task is instantiated, it will first use any argument as the value of the parameter, eg. if you instantiate a = TaskA(x=44) then a.x == 44. When the value is not provided, the value will be resolved in this order of falling priority:

• Any value provided on the command line:

  • To the root task (eg. --param xyz)

  • Then to the class, using the qualified task name syntax (eg. --TaskA-param xyz).

• With [TASK_NAME]>PARAM_NAME: <serialized value> syntax. See Parameters from config Ingestion

• Any default value set using the default flag.

Parameter objects may be reused, but you must then set the positional=False flag.

ping_interval

Parameter whose value is a float.

keep_alive

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

count_uniques

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

count_last_scheduled

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

wait_interval

Parameter whose value is a float.

wait_jitter

Parameter whose value is a float.

max_keep_alive_idle_duration

Class that maps to timedelta using strings in any of the following forms:

• A bare number is interpreted as duration in seconds.

• n {w[eek[s]]|d[ay[s]]|h[our[s]]|m[inute[s]|s[second[s]]} (e.g. “1 week 2 days” or “1 h”)

  Note: multiple arguments must be supplied in longest to shortest unit order

• ISO 8601 duration PnDTnHnMnS (each field optional, years and months not supported)

• ISO 8601 duration PnW

See https://en.wikipedia.org/wiki/ISO_8601#Durations

max_reschedules

Parameter whose value is an int.

timeout

Parameter whose value is an int.

task_limit

Parameter whose value is an int.

retry_external_tasks

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

send_failure_email

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

no_install_shutdown_handler

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

check_unfulfilled_deps

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

check_complete_on_run

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

force_multiprocessing

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

task_process_context

Class to parse optional parameters.

cache_task_completion

A Parameter whose value is a bool. This parameter has an implicit default value of False. For the command line interface this means that the value is False unless you add "--the-bool-parameter" to your command without giving a parameter value. This is considered implicit parsing (the default). However, in some situations one might want to give the explicit bool value ("--the-bool-parameter true|false"), e.g. when you configure the default value to be True. This is called explicit parsing. When omitting the parameter value, it is still considered True but to avoid ambiguities during argument parsing, make sure to always place bool parameters behind the task family on the command line when using explicit parsing.

You can toggle between the two parsing modes on a per-parameter base via

class MyTask(luigi.Task):
    implicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.IMPLICIT_PARSING)
    explicit_bool = luigi.BoolParameter(parsing=luigi.BoolParameter.EXPLICIT_PARSING)

or globally by

luigi.BoolParameter.parsing = luigi.BoolParameter.EXPLICIT_PARSING

for all bool parameters instantiated after this line.

class luigi.worker.KeepAliveThread(scheduler, worker_id, ping_interval, rpc_message_callback)

Periodically tell the scheduler that the worker still lives.

This constructor should always be called with keyword arguments. Arguments are:

group should be None; reserved for future extension when a ThreadGroup class is implemented.

target is the callable object to be invoked by the run() method. Defaults to None, meaning nothing is called.

name is the thread name. By default, a unique name is constructed of the form “Thread-N” where N is a small decimal number.

args is a list or tuple of arguments for the target invocation. Defaults to ().

kwargs is a dictionary of keyword arguments for the target invocation. Defaults to {}.

If a subclass overrides the constructor, it must make sure to invoke the base class constructor (Thread.__init__()) before doing anything else to the thread.

stop()

run()

Method representing the thread’s activity.

You may override this method in a subclass. The standard run() method invokes the callable object passed to the object’s constructor as the target argument, if any, with sequential and keyword arguments taken from the args and kwargs arguments, respectively.

luigi.worker.rpc_message_callback(fn)

class luigi.worker.Worker(scheduler=None, worker_id=None, worker_processes=1, assistant=False, **kwargs)

Worker object communicates with a scheduler.

Simple class that talks to a scheduler and:

• tells the scheduler what it has to do + its dependencies

• asks for stuff to do (pulls it in a loop and runs it)

add(task, multiprocess=False, processes=0)

Add a Task for the worker to check and possibly schedule and run.

Returns True if task and its dependencies were successfully scheduled or completed before.

handle_interrupt(signum, _)

Stops the assistant from asking for more work on SIGUSR1

run()

Returns True if all scheduled tasks were executed successfully.

set_worker_processes(n)

dispatch_scheduler_message(task_id, message_id, content, **kwargs)