Skip to content

infrastructure.cache.cache_metrics

src.infrastructure.cache.cache_metrics

Cache metrics tracking for observability.

This module provides lightweight in-memory metrics tracking for cache operations. Metrics include hit/miss counts and hit rates for monitoring cache effectiveness.

Usage

from src.infrastructure.cache.cache_metrics import CacheMetrics

metrics = CacheMetrics() metrics.record_hit("user") metrics.record_miss("user")

stats = metrics.get_stats("user") print(f"Hit rate: {stats['hit_rate']:.2%}")

Classes

CacheStats dataclass

Cache statistics for a specific cache namespace.

Attributes:

Name Type Description
hits int

Number of cache hits (value found in cache).
misses int

Number of cache misses (value not in cache).
errors int

Number of cache operation errors.
Source code in src/infrastructure/cache/cache_metrics.py 
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
@dataclass
class CacheStats:
    """Cache statistics for a specific cache namespace.

    Attributes:
        hits: Number of cache hits (value found in cache).
        misses: Number of cache misses (value not in cache).
        errors: Number of cache operation errors.
    """

    hits: int = 0
    misses: int = 0
    errors: int = 0

    @property
    def total_requests(self) -> int:
        """Total cache requests (hits + misses)."""
        return self.hits + self.misses

    @property
    def hit_rate(self) -> float:
        """Cache hit rate (0.0 to 1.0)."""
        if self.total_requests == 0:
            return 0.0
        return self.hits / self.total_requests

    def to_dict(self) -> dict[str, Any]:
        """Convert stats to dictionary for JSON serialization."""
        return {
            "hits": self.hits,
            "misses": self.misses,
            "errors": self.errors,
            "total_requests": self.total_requests,
            "hit_rate": round(self.hit_rate, 4),
        }
Attributes
total_requests property 
total_requests: int

Total cache requests (hits + misses).

hit_rate property 
hit_rate: float

Cache hit rate (0.0 to 1.0).

Functions
to_dict 
to_dict() -> dict[str, Any]

Convert stats to dictionary for JSON serialization.

Source code in src/infrastructure/cache/cache_metrics.py 
49
50
51
52
53
54
55
56
57
def to_dict(self) -> dict[str, Any]:
    """Convert stats to dictionary for JSON serialization."""
    return {
        "hits": self.hits,
        "misses": self.misses,
        "errors": self.errors,
        "total_requests": self.total_requests,
        "hit_rate": round(self.hit_rate, 4),
    }

CacheMetrics

In-memory cache metrics tracker.

Thread-safe metrics tracking for cache operations. Tracks hits, misses, and errors per cache namespace (e.g., "user", "provider", "accounts").

This is a lightweight, in-memory tracker suitable for development and production monitoring. For production, metrics can be exported to observability platforms (Prometheus, Datadog, etc.).

Example

metrics = CacheMetrics()

Record operations

metrics.record_hit("user") metrics.record_miss("provider") metrics.record_error("accounts")

Get stats

stats = metrics.get_stats("user") print(f"User cache hit rate: {stats['hit_rate']:.2%}")

Get all stats

all_stats = metrics.get_all_stats()

Source code in src/infrastructure/cache/cache_metrics.py 
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
class CacheMetrics:
    """In-memory cache metrics tracker.

    Thread-safe metrics tracking for cache operations. Tracks hits, misses,
    and errors per cache namespace (e.g., "user", "provider", "accounts").

    This is a lightweight, in-memory tracker suitable for development and
    production monitoring. For production, metrics can be exported to
    observability platforms (Prometheus, Datadog, etc.).

    Example:
        metrics = CacheMetrics()

        # Record operations
        metrics.record_hit("user")
        metrics.record_miss("provider")
        metrics.record_error("accounts")

        # Get stats
        stats = metrics.get_stats("user")
        print(f"User cache hit rate: {stats['hit_rate']:.2%}")

        # Get all stats
        all_stats = metrics.get_all_stats()
    """

    def __init__(self) -> None:
        """Initialize metrics tracker with empty counters."""
        self._stats: dict[str, CacheStats] = defaultdict(CacheStats)
        self._lock = Lock()

    def record_hit(self, namespace: str) -> None:
        """Record a cache hit.

        Args:
            namespace: Cache namespace (e.g., "user", "provider").
        """
        with self._lock:
            self._stats[namespace].hits += 1

    def record_miss(self, namespace: str) -> None:
        """Record a cache miss.

        Args:
            namespace: Cache namespace (e.g., "user", "provider").
        """
        with self._lock:
            self._stats[namespace].misses += 1

    def record_error(self, namespace: str) -> None:
        """Record a cache operation error.

        Args:
            namespace: Cache namespace (e.g., "user", "provider").
        """
        with self._lock:
            self._stats[namespace].errors += 1

    def get_stats(self, namespace: str) -> dict[str, Any]:
        """Get statistics for a specific namespace.

        Args:
            namespace: Cache namespace to query.

        Returns:
            Dictionary with hits, misses, errors, total_requests, hit_rate.
        """
        with self._lock:
            stats = self._stats.get(namespace, CacheStats())
            return stats.to_dict()

    def get_all_stats(self) -> dict[str, dict[str, Any]]:
        """Get statistics for all namespaces.

        Returns:
            Dictionary mapping namespace to stats dictionary.
        """
        with self._lock:
            return {
                namespace: stats.to_dict() for namespace, stats in self._stats.items()
            }

    def reset(self, namespace: str | None = None) -> None:
        """Reset metrics.

        Args:
            namespace: Optional namespace to reset. If None, reset all.
        """
        with self._lock:
            if namespace is None:
                self._stats.clear()
            else:
                self._stats[namespace] = CacheStats()
Functions
__init__ 
__init__() -> None
Source code in src/infrastructure/cache/cache_metrics.py 
86
87
88
89
def __init__(self) -> None:
    """Initialize metrics tracker with empty counters."""
    self._stats: dict[str, CacheStats] = defaultdict(CacheStats)
    self._lock = Lock()
record_hit 
record_hit(namespace: str) -> None

Record a cache hit.

Parameters:

Name Type Description Default
namespace str

Cache namespace (e.g., "user", "provider").

 required
Source code in src/infrastructure/cache/cache_metrics.py 
91
92
93
94
95
96
97
98
def record_hit(self, namespace: str) -> None:
    """Record a cache hit.

    Args:
        namespace: Cache namespace (e.g., "user", "provider").
    """
    with self._lock:
        self._stats[namespace].hits += 1
record_miss 
record_miss(namespace: str) -> None

Record a cache miss.

Parameters:

Name Type Description Default
namespace str

Cache namespace (e.g., "user", "provider").

 required
Source code in src/infrastructure/cache/cache_metrics.py 
100
101
102
103
104
105
106
107
def record_miss(self, namespace: str) -> None:
    """Record a cache miss.

    Args:
        namespace: Cache namespace (e.g., "user", "provider").
    """
    with self._lock:
        self._stats[namespace].misses += 1
record_error 
record_error(namespace: str) -> None

Record a cache operation error.

Parameters:

Name Type Description Default
namespace str

Cache namespace (e.g., "user", "provider").

 required
Source code in src/infrastructure/cache/cache_metrics.py 
109
110
111
112
113
114
115
116
def record_error(self, namespace: str) -> None:
    """Record a cache operation error.

    Args:
        namespace: Cache namespace (e.g., "user", "provider").
    """
    with self._lock:
        self._stats[namespace].errors += 1
get_stats 
get_stats(namespace: str) -> dict[str, Any]

Get statistics for a specific namespace.

Parameters:

Name Type Description Default
namespace str

Cache namespace to query.

 required

Returns:

Type Description
dict[str, Any]

Dictionary with hits, misses, errors, total_requests, hit_rate.
Source code in src/infrastructure/cache/cache_metrics.py 
118
119
120
121
122
123
124
125
126
127
128
129
def get_stats(self, namespace: str) -> dict[str, Any]:
    """Get statistics for a specific namespace.

    Args:
        namespace: Cache namespace to query.

    Returns:
        Dictionary with hits, misses, errors, total_requests, hit_rate.
    """
    with self._lock:
        stats = self._stats.get(namespace, CacheStats())
        return stats.to_dict()
get_all_stats 
get_all_stats() -> dict[str, dict[str, Any]]

Get statistics for all namespaces.

Returns:

Type Description
dict[str, dict[str, Any]]

Dictionary mapping namespace to stats dictionary.
Source code in src/infrastructure/cache/cache_metrics.py 
131
132
133
134
135
136
137
138
139
140
def get_all_stats(self) -> dict[str, dict[str, Any]]:
    """Get statistics for all namespaces.

    Returns:
        Dictionary mapping namespace to stats dictionary.
    """
    with self._lock:
        return {
            namespace: stats.to_dict() for namespace, stats in self._stats.items()
        }
reset 
reset(namespace: str | None = None) -> None

Reset metrics.

Parameters:

Name Type Description Default
namespace str | None

Optional namespace to reset. If None, reset all.

 None
Source code in src/infrastructure/cache/cache_metrics.py 
142
143
144
145
146
147
148
149
150
151
152
def reset(self, namespace: str | None = None) -> None:
    """Reset metrics.

    Args:
        namespace: Optional namespace to reset. If None, reset all.
    """
    with self._lock:
        if namespace is None:
            self._stats.clear()
        else:
            self._stats[namespace] = CacheStats()

Functions

get_cache_metrics 

get_cache_metrics() -> CacheMetrics

Get global cache metrics singleton.

Returns:

Type Description
CacheMetrics

Global CacheMetrics instance.
Source code in src/infrastructure/cache/cache_metrics.py 
159
160
161
162
163
164
165
166
167
168
def get_cache_metrics() -> CacheMetrics:
    """Get global cache metrics singleton.

    Returns:
        Global CacheMetrics instance.
    """
    global _metrics_instance
    if _metrics_instance is None:
        _metrics_instance = CacheMetrics()
    return _metrics_instance