sentry_arroyo 2.39.1

A library for working with streaming data.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

from typing import MutableSequence, NamedTuple, Optional, Union

from arroyo.utils.metrics import MetricName, Metrics, Tags


class Increment(NamedTuple):
    name: MetricName
    value: Union[int, float]
    tags: Optional[Tags]


class Gauge(NamedTuple):
    name: MetricName
    value: Union[int, float]
    tags: Optional[Tags]


class Timing(NamedTuple):
    name: MetricName
    value: Union[int, float]
    tags: Optional[Tags]


MetricCall = Union[Increment, Gauge, Timing]


class _TestingMetricsBackend(Metrics):
    """
    A metrics backend that logs all metrics recorded. Intended for testing
    the behavior of the metrics implementations themselves. Not for general
    use (it will cause unbounded memory consumption.)
    """

    def __init__(self) -> None:
        self.calls: MutableSequence[Union[Increment, Gauge, Timing]] = []

    def increment(
        self,
        name: MetricName,
        value: Union[int, float] = 1,
        tags: Optional[Tags] = None,
    ) -> None:
        self.calls.append(Increment(name, value, tags))

    def gauge(
        self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
    ) -> None:
        self.calls.append(Gauge(name, value, tags))

    def timing(
        self, name: MetricName, value: Union[int, float], tags: Optional[Tags] = None
    ) -> None:
        self.calls.append(Timing(name, value, tags))


TestingMetricsBackend = _TestingMetricsBackend()