8000 make counter classes public by vittoriopolverino · Pull Request #12386 · localstack/localstack · GitHub
[go: up one dir, main page]
Skip to content

make counter classes public #12386

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 4 commits into from
Mar 14, 2025
Merged

make counter classes public #12386

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
915579f
make counter classes public allowing to use them for typing
vittoriopolverino Mar 13, 2025
22aadcf
Update localstack-core/localstack/utils/analytics/metrics.py
vittoriopolverino Mar 13, 2025
8bc1c26
chore: add docstring to clarify LabeledCounterMetric, CounterMetric, …
vittoriopolverino Mar 13, 2025
68e55e4
chore: add docstring to clarify LabeledCounterMetric, CounterMetric, …
vittoriopolverino Mar 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 21 additions & 18 deletions localstack-core/localstack/utils/analytics/metrics.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,16 +99,17 @@ def collect(self) -> List[Dict[str, Union[str, int]]]:
pass


< 8000 /td> class _Counter:
class BaseCounter:
"""
A thread-safe counter for any kind of tracking.
This class should not be instantiated directly, use the Counter class instead.
"""

_mutex: threading.Lock
_count: int

def __init__(self):
super(_Counter, self).__init__()
super(BaseCounter, self).__init__()
self._mutex = threading.Lock()
self._count = 0

Expand Down Expand Up @@ -136,17 +137,18 @@ def reset(self) -> None:
self._count = 0


class _CounterMetric(Metric, _Counter):
class CounterMetric(Metric, BaseCounter):
"""
A thread-safe counter for tracking occurrences of an event without labels.
This class should not be instantiated directly, use the Counter class instead.
"""

_namespace: Optional[str]
_type: str

def __init__(self, name: str, namespace: Optional[str] = ""):
Metric.__init__(self, name=name)
_Counter.__init__(self)
BaseCounter.__init__(self)

self._namespace = namespace.strip() if namespace else ""
self._type = "counter"
Expand All @@ -170,20 +172,21 @@ def collect(self) -> List[Dict[str, Union[str, int]]]:
]


class _LabeledCounterMetric(Metric):
class LabeledCounterMetric(Metric):
"""
A labeled counter that tracks occurrences of an event across different label combinations.
This class should not be instantiated directly, use the Counter class instead.
"""

_namespace: Optional[str]
_type: str
_unit: str
_labels: list[str]
_label_values: Tuple[Optional[Union[str, float]], ...]
_counters_by_label_values: defaultdict[Tuple[Optional[Union[str, float]], ...], _Counter]
_counters_by_label_values: defaultdict[Tuple[Optional[Union[str, float]], ...], BaseCounter]

def __init__(self, name: str, labels: List[str], namespace: Optional[str] = ""):
super(_LabeledCounterMetric, self).__init__(name=name)
super(LabeledCounterMetric, self).__init__(name=name)

if not labels:
raise ValueError("At least one label is required; the labels list cannot be empty.")
Expand All @@ -197,15 +200,15 @@ def __init__(self, name: str, labels: List[str], namespace: Optional[str] = ""):
self._namespace = namespace.strip() if namespace else ""
self._type = "counter"
self._labels = labels
self._counters_by_label_values = defaultdict(_Counter)
self._counters_by_label_values = defaultdict(BaseCounter)
MetricRegistry().register(self)

def labels(self, **kwargs: Union[str, float, None]) -> _Counter:
def labels(self, **kwargs: Union[str, float, None]) -> BaseCounter:
"""
Create a scoped counter instance with specific label values.

This method assigns values to the predefined labels of a labeled counter and returns
a _Counter object (``) that allows tracking metrics for that specific
a BaseCounter object that allows tracking metrics for that specific
combination of label values.

:raises ValueError:
Expand Down Expand Up @@ -260,26 +263,26 @@ class Counter:
A factory class for creating counter instances.

This class provides a flexible way to create either a simple counter
(`_SimpleCounter`) or a labeled counter (`_LabeledCounter`) based on
(`CounterMetric`) or a labeled counter (`LabeledCounterMetric`) based on
whether labels are provided.
"""

@overload
def __new__(cls, name: str, namespace: Optional[str] = "") -> _CounterMetric:
return _CounterMetric(namespace=namespace, name=name)
def __new__(cls, name: str, namespace: Optional[str] = "") -> CounterMetric:
return CounterMetric(namespace=namespace, name=name)

@overload
def __new__(
cls, name: str, labels: List[str], namespace: Optional[str] = ""
) -> _LabeledCounterMetric:
return _LabeledCounterMetric(namespace=namespace, name=name, labels=labels)
) -> LabeledCounterMetric:
return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)

def __new__(
cls, name: str, namespace: Optional[str] = "", labels: Optional[List[str]] = None
) -> Union[_CounterMetric, _LabeledCounterMetric]:
) -> Union[CounterMetric, LabeledCounterMetric]:
if labels is not None:
return _LabeledCounterMetric(namespace=namespace, name=name, labels=labels)
return _CounterMetric(namespace=namespace, name=name)
return LabeledCounterMetric(namespace=namespace, name=name, labels=labels)
return CounterMetric(namespace=namespace, name=name)


@hooks.on_infra_shutdown()
Expand Down
Loading
0