Source code for luigi.metrics

import abc

from enum import Enum
from luigi import six


[docs]class MetricsCollectors(Enum): default = 1 none = 1 datadog = 2 @classmethod def get(cls, which): if which == MetricsCollectors.none: return NoMetricsCollector() elif which == MetricsCollectors.datadog: from luigi.contrib.datadog_metric import DatadogMetricsCollector return DatadogMetricsCollector() else: raise ValueError("MetricsCollectors value ' {0} ' isn't supported", which)
[docs]@six.add_metaclass(abc.ABCMeta) class MetricsCollector(object): """Abstractable MetricsCollector base class that can be replace by tool specific implementation. """ @abc.abstractmethod def __init__(self): pass
[docs] @abc.abstractmethod def handle_task_started(self, task): pass
[docs] @abc.abstractmethod def handle_task_failed(self, task): pass
[docs] @abc.abstractmethod def handle_task_disabled(self, task, config): pass
[docs] @abc.abstractmethod def handle_task_done(self, task): pass
[docs]class NoMetricsCollector(MetricsCollector): """Empty MetricsCollector when no collector is being used """ def __init__(self): pass
[docs] def handle_task_started(self, task): pass
[docs] def handle_task_failed(self, task): pass
[docs] def handle_task_disabled(self, task, config): pass
[docs] def handle_task_done(self, task): pass