Class Enum

prometheus_client/metrics.py:744–802

Enum metric, which of a set of states is true. Example usage: from prometheus_client import Enum e = Enum('task_state', 'Description of enum', states=['starting', 'running', 'stopped']) e.state('running') The first listed state will be the default.

class Enum(MetricWrapperBase):
    """Enum metric, which of a set of states is true.

     Example usage:
        from prometheus_client import Enum

        e = Enum('task_state', 'Description of enum',
          states=['starting', 'running', 'stopped'])
        e.state('running')

     The first listed state will be the default.
     Enum metrics do not work in multiprocess mode.
    """
    _type = 'stateset'

    def __init__(self,
                 name: str,
                 documentation: str,
                 labelnames: Sequence[str] = (),
                 namespace: str = '',
                 subsystem: str = '',
                 unit: str = '',
                 registry: Optional[CollectorRegistry] = REGISTRY,
                 _labelvalues: Optional[Sequence[str]] = None,
                 states: Optional[Sequence[str]] = None,
                 ):
        super().__init__(
            name=name,
            documentation=documentation,
            labelnames=labelnames,
            namespace=namespace,
            subsystem=subsystem,
            unit=unit,
            registry=registry,
            _labelvalues=_labelvalues,
        )
        if name in labelnames:
            raise ValueError(f'Overlapping labels for Enum metric: {name}')
        if not states:
            raise ValueError(f'No states provided for Enum metric: {name}')
        self._kwargs['states'] = self._states = states

    def _metric_init(self) -> None:
        self._value = 0
        self._lock = Lock()

    def state(self, state: str) -> None:
        """Set enum metric state."""
        self._raise_if_not_observable()
        with self._lock:
            self._value = self._states.index(state)

    def _child_samples(self) -> Iterable[Sample]:
        with self._lock:
            return [
                Sample('', {self._name: s}, 1 if i == self._value else 0, None, None)
                for i, s
                in enumerate(self._states)