Luigi Monitor

1.1.4 verified Sat May 09 auth: no python

Luigi Monitor sends summary messages of your Luigi pipeline runs to Slack (via webhook). It provides a lightweight wrapper to augment Luigi's scheduling events and report success/failure in real-time. Current version 1.1.4, with no recent releases; the project is in maintenance mode.

pip install luigi-monitor

Common errors

error ImportError: cannot import name 'LuigiMonitor' ↓

cause Incorrect import path; often using hyphen instead of underscore.

fix

Install with pip install luigi-monitor, then import as 'from luigi_monitor import LuigiMonitor'

error AttributeError: module 'luigi_monitor' has no attribute 'LuigiMonitor' ↓

cause Old version of the library (pre-1.0) used a different API; likely installed wrong package or outdated version.

fix

Upgrade to latest: pip install --upgrade luigi-monitor

Warnings

gotcha The import path uses an underscore (luigi_monitor) despite the PyPI package name having a hyphen. Attempting to import luigi-monitor directly will fail. ↓

fix Use 'from luigi_monitor import LuigiMonitor'

gotcha Slack webhook URL must be provided either via constructor or SLACK_URL env var. If missing, monitor silently falls back to no-op without warning. ↓

fix Ensure SLACK_WEBHOOK_URL environment variable is set, or pass slack_url explicitly.

deprecated The project has had no updates since 2019; may not work with Luigi 3.x due to internal event API changes. No official compatibility statement. ↓

fix Test with your Luigi version; consider forking or switching to alternative monitoring.

Imports

LuigiMonitor

wrong

from luigimonitor import LuigiMonitor

correct

from luigi_monitor import LuigiMonitor

Module name has hyphen, not underscore in import path

Quickstart

Wrap Luigi tasks with @monitor.notify to send Slack messages on completion/failure.

from luigi_monitor import LuigiMonitor
import os

monitor = LuigiMonitor(
    slack_url=os.environ.get('SLACK_WEBHOOK_URL', ''),
    channel='#pipeline-alerts',
    username='luigi-bot',
    icon_emoji=':robot_face:'
)

@monitor.notify
class MyTask(luigi.Task):
    def run(self):
        pass

if __name__ == '__main__':
    luigi.run()