Luigi Workflow Management

Warnings

• breaking Luigi 3.x series requires Python >= 3.10 and < 3.14. Projects on older Python 3 versions (e.g., 3.6-3.9) or newer versions (3.14+) will not be compatible.
  Fix: Upgrade your Python environment to a compatible version (3.10, 3.11, 3.12, 3.13) before upgrading Luigi to 3.x.
• gotcha Luigi tasks require a scheduler to run. By default, it looks for a running `luigid` daemon. For simple local execution or development, remember to use the `--local-scheduler` flag when running tasks from the command line, or `local_scheduler=True` when using `luigi.build()` programmatically.
  Fix: Start `luigid` in a separate terminal (`luigid --port 8082`) or always include `--local-scheduler` in your command line execution or `local_scheduler=True` in `luigi.build()` calls.
• deprecated The `pkg_resources` library, which was historically used for introspection and entry points, has been removed in Luigi 3.7.3. While this is primarily an internal change, it might affect custom plugins or integrations that directly or indirectly relied on Luigi's usage of `pkg_resources`.
  Fix: Review custom plugins or extensions for any direct calls or implicit dependencies on `pkg_resources` and migrate to `importlib.metadata` (for Python 3.8+) or alternative methods.
• gotcha Luigi determines task completion solely based on whether the `output()` targets `exist()`. If a task's `run()` method completes but fails to create its defined `output()`, Luigi will consider it incomplete and re-run it in subsequent invocations. Similarly, manually deleting an output file will cause Luigi to re-run the producing task.
  Fix: Ensure that every `run()` method reliably creates the target specified by its `output()` method. For tasks that don't produce physical outputs, consider using a dummy `LocalTarget('/tmp/luigi_flag_file')` or `luigi.MockTarget` for testing, or ensure their dependencies cover the actual work.

Install

• pip install luigi Install Luigi

Imports

• Task

  import luigi
  
  class MyTask(luigi.Task): ...

• Parameter

  import luigi
  
  class MyTask(luigi.Task):
      param = luigi.Parameter()

• LocalTarget

  import luigi
  
  luigi.LocalTarget('path/to/file')

• build

  import luigi
  
  luigi.build([my_task_instance], local_scheduler=True)

Quickstart

This quickstart defines a simple Luigi task `GenerateReport` that takes a `date` parameter, simulates generating a report, and writes it to a local file. It demonstrates defining a task, its `run` method, and its `output` target using `luigi.LocalTarget`. The example also shows how to trigger tasks programmatically using `luigi.build` with a local scheduler.

import luigi
import datetime
import os

class GenerateReport(luigi.Task):
    date = luigi.DateParameter(default=datetime.date.today())

    def run(self):
        # Simulate some data processing
        report_content = f"Daily Report for {self.date.isoformat()}\n" \
                         f"Generated by Luigi.\n"

        # Write the report to a target file
        with self.output().open('w') as f:
            f.write(report_content)

    def output(self):
        # Define where the output of this task will be stored
        # Using an environment variable for flexibility or defaulting to current dir
        output_dir = os.environ.get('LUIGI_OUTPUT_DIR', '.')
        return luigi.LocalTarget(os.path.join(output_dir, f'report_{self.date.isoformat()}.txt'))

if __name__ == '__main__':
    # To run this task using the command line (most common):
    # 1. Start the Luigi scheduler daemon in a separate terminal: luigid --port 8082
    # 2. Run your script: python your_script_name.py GenerateReport --date 2023-10-26
    #    (or omit --date for today's date if default is set)
    # 3. If you don't want to run luigid, use the --local-scheduler flag:
    #    python your_script_name.py GenerateReport --local-scheduler

    # For programmatic execution (e.g., in a wrapper script or test):
    # The 'local_scheduler=True' ensures it runs without an external luigid daemon.
    luigi.build([GenerateReport(date=datetime.date(2023, 10, 26))], local_scheduler=True)