luigi.server

Simple REST server that takes commands in a JSON payload Interface to the Scheduler class. See Using the Central Scheduler for more info.

Functions

app(scheduler)
from_utc(utcTime[, fmt])	convert UTC time string to time.struct_time: change datetime.datetime to time, return time.struct_time type
run([api_port, address, unix_socket, scheduler])	Runs one instance of the API server.
stop()

Classes

AllRunHandler(application, request, **kwargs)
BaseTaskHistoryHandler(application, request, ...)
ByIdHandler(application, request, **kwargs)
ByNameHandler(application, request, **kwargs)
ByParamsHandler(application, request, **kwargs)
ByTaskIdHandler(application, request, **kwargs)
MetricsHandler(application, request, **kwargs)
RPCHandler(*args, **kwargs)	Handle remote scheduling calls using rpc.RemoteSchedulerResponder.
RecentRunHandler(application, request, **kwargs)
RootPathHandler(application, request, **kwargs)
SelectedRunHandler(application, request, ...)
cors(*args, **kwargs)

class luigi.server.cors(*args, **kwargs)

enabled

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.

allow_any_origin

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.

allow_null_origin

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.

max_age

Parameter whose value is an int.

allowed_methods

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.

allowed_headers

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.

exposed_headers

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.

allow_credentials

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.

allowed_origins

Parameter whose value is a list.

In the task definition, use

class MyTask(luigi.Task):
  grades = luigi.ListParameter()

    def run(self):
        sum = 0
        for element in self.grades:
            sum += element
        avg = sum / len(self.grades)

At the command line, use

$ luigi --module my_tasks MyTask --grades <JSON string>

Simple example with two grades:

$ luigi --module my_tasks MyTask --grades '[100,70]'

It is possible to provide a JSON schema that should be validated by the given value:

class MyTask(luigi.Task):
  grades = luigi.ListParameter(
    schema={
      "type": "array",
      "items": {
        "type": "number",
        "minimum": 0,
        "maximum": 10
      },
      "minItems": 1
    }
  )

  def run(self):
        sum = 0
        for element in self.grades:
            sum += element
        avg = sum / len(self.grades)

Using this schema, the following command will work:

$ luigi --module my_tasks MyTask --numbers '[1, 8.7, 6]'

while these commands will fail because the parameter is not valid:

$ luigi --module my_tasks MyTask --numbers '[]'  # must have at least 1 element
$ luigi --module my_tasks MyTask --numbers '[-999, 999]'  # elements must be in [0, 10]

Finally, the provided schema can be a custom validator:

custom_validator = jsonschema.Draft4Validator(
  schema={
    "type": "array",
    "items": {
      "type": "number",
      "minimum": 0,
      "maximum": 10
    },
    "minItems": 1
  }
)

class MyTask(luigi.Task):
  grades = luigi.ListParameter(schema=custom_validator)

  def run(self):
        sum = 0
        for element in self.grades:
            sum += element
        avg = sum / len(self.grades)

class luigi.server.RPCHandler(*args, **kwargs)

Handle remote scheduling calls using rpc.RemoteSchedulerResponder.

initialize(scheduler)

options(*args)

get(method)

post(method)

class luigi.server.BaseTaskHistoryHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

initialize(scheduler)

get_template_path()

Override to customize template path for each handler.

By default, we use the template_path application setting. Return None to load templates relative to the calling file.

class luigi.server.AllRunHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get()

class luigi.server.SelectedRunHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get(name)

luigi.server.from_utc(utcTime, fmt=None)

convert UTC time string to time.struct_time: change datetime.datetime to time, return time.struct_time type

class luigi.server.RecentRunHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get()

class luigi.server.ByNameHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get(name)

class luigi.server.ByIdHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get(id)

class luigi.server.ByTaskIdHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get(task_id)

class luigi.server.ByParamsHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get(name)

class luigi.server.RootPathHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

get()

head()

HEAD endpoint for health checking the scheduler

class luigi.server.MetricsHandler(application: Application, request: HTTPServerRequest, **kwargs: Any)

initialize(scheduler)

get()

luigi.server.app(scheduler)

luigi.server.run(api_port=8082, address=None, unix_socket=None, scheduler=None)

Runs one instance of the API server.

luigi.server.stop()