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
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
# Copyright (c) 2022 Tulir Asokan
#
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
from __future__ import annotations
from typing import Any, cast
class _NoopPrometheusEntity:
"""NoopPrometheusEntity is a class that can be used as a no-op placeholder for prometheus
metrics objects when prometheus_client isn't installed."""
def __init__(self, *args, **kwargs):
pass
def __call__(self, *args, **kwargs):
if not kwargs and len(args) == 1 and callable(args[0]):
return args[0]
return self
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
pass
def __getattr__(self, item):
return self
try:
from prometheus_client import Counter, Enum, Gauge, Histogram, Info, Summary
is_installed = True
except ImportError:
Counter = Gauge = Summary = Histogram = Info = Enum = cast(Any, _NoopPrometheusEntity)
is_installed = False
def async_time(metric: Gauge | Summary | Histogram):
"""
Measure the time that each execution of the decorated async function takes.
This is equivalent to the ``time`` method-decorator in the metrics, but
supports async functions.
Args:
metric: The metric instance to store the measures in.
"""
if not hasattr(metric, "time") or not callable(metric.time):
raise ValueError("async_time only supports metrics that support timing")
def decorator(fn):
async def wrapper(*args, **kwargs):
with metric.time():
return await fn(*args, **kwargs)
return wrapper if is_installed else fn
return decorator
__all__ = [
"Counter",
"Gauge",
"Summary",
"Histogram",
"Info",
"Enum",
"async_time",
"is_installed",
]
|
