Common - Method Wrappers:

import grpc, json, logging, threading

from enum import Enum

from prettytable import PrettyTable

from typing import Any, Dict, List, Set, Tuple

from typing import Any, Dict, List, Optional, Set, Tuple

from prometheus_client import Counter, Histogram

from prometheus_client.metrics import MetricWrapperBase, INF

from common.tools.grpc.Tools import grpc_message_to_json_string

            return MetricsPool.metrics[metric_name]

    def get_metrics(

        self, method : str

    ) -> Tuple[MetricWrapperBase, MetricWrapperBase, MetricWrapperBase, MetricWrapperBase]:

        self, method : str, labels : Optional[Dict[str, str]] = None

    ) -> Tuple[Histogram, Counter, Counter, Counter]:

        histogram_duration : Histogram = self.get_or_create(method, MetricTypeEnum.HISTOGRAM_DURATION)

        counter_started    : Counter   = self.get_or_create(method, MetricTypeEnum.COUNTER_STARTED)

        counter_completed  : Counter   = self.get_or_create(method, MetricTypeEnum.COUNTER_COMPLETED)

        counter_failed     : Counter   = self.get_or_create(method, MetricTypeEnum.COUNTER_FAILED)

        if len(self._labels) > 0:

            histogram_duration = histogram_duration.labels(**(self._labels))

            counter_started    = counter_started.labels(**(self._labels))

            counter_completed  = counter_completed.labels(**(self._labels))

            counter_failed     = counter_failed.labels(**(self._labels))

        if labels is None and len(self._labels) > 0:

            labels = self._labels

        if labels is not None and len(labels) > 0:

            histogram_duration = histogram_duration.labels(**labels)

            counter_started    = counter_started.labels(**labels)

            counter_completed  = counter_completed.labels(**labels)

            counter_failed     = counter_failed.labels(**labels)

        return histogram_duration, counter_started, counter_completed, counter_failed