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
    prometheus = 3

[docs]    @classmethod
    def get(cls, which):
        if which == MetricsCollectors.none:
            return NoMetricsCollector()
        elif which == MetricsCollectors.datadog:
            from luigi.contrib.datadog_metric import DatadogMetricsCollector
            return DatadogMetricsCollector()
        elif which == MetricsCollectors.prometheus:
            from luigi.contrib.prometheus_metric import PrometheusMetricsCollector
            return PrometheusMetricsCollector()
        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]    def generate_latest(self):
        return

[docs]    def configure_http_handler(self, http_handler):
        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