Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1import threading
2from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
4# Optional import - OTel SDK may not be installed
5# Use Any as fallback type when OTel is not available
6if TYPE_CHECKING:
7 try:
8 from opentelemetry.metrics import Observation
9 except ImportError:
10 Observation = Any # type: ignore[misc]
11else:
12 Observation = Any
15class ObservablesRegistry:
16 """
17 Global registry for storing callbacks for observable metrics.
18 """
20 def __init__(self, registry: Dict[str, List[Callable[[], List[Any]]]] = None):
21 self._registry = registry or {}
22 self._lock = threading.Lock()
24 def register(self, name: str, callback: Callable[[], List[Any]]) -> None:
25 """
26 Register a callback for an observable metric.
27 """
28 with self._lock:
29 self._registry.setdefault(name, []).append(callback)
31 def get(self, name: str) -> List[Callable[[], List[Any]]]:
32 """
33 Get all callbacks for an observable metric.
34 """
35 with self._lock:
36 return self._registry.get(name, [])
38 def clear(self) -> None:
39 """
40 Clear the registry.
41 """
42 with self._lock:
43 self._registry.clear()
45 def __len__(self) -> int:
46 """
47 Get the number of registered callbacks.
48 """
49 return len(self._registry)
52# Global singleton instance
53_observables_registry_instance: Optional[ObservablesRegistry] = None
56def get_observables_registry_instance() -> ObservablesRegistry:
57 """
58 Get the global observables registry singleton instance.
60 This is the Pythonic way to get the singleton instance.
62 Returns:
63 The global ObservablesRegistry singleton
65 Example:
66 >>>
67 >>> registry = get_observables_registry_instance()
68 >>> registry.register('my_metric', my_callback)
69 """
70 global _observables_registry_instance
72 if _observables_registry_instance is None:
73 _observables_registry_instance = ObservablesRegistry()
75 return _observables_registry_instance