Skip to content

honeyhive.utils.cache

Caching utilities for HoneyHive.

CacheConfig dataclass

Configuration for cache.

Source code in src/honeyhive/utils/cache.py 
10
11
12
13
14
15
16
17
@dataclass
class CacheConfig:
    """Configuration for cache."""

    max_size: int = 1000
    default_ttl: float = 300.0  # 5 minutes
    cleanup_interval: float = 60.0  # 1 minute
    enable_stats: bool = True

max_size class-attribute instance-attribute 

max_size: int = 1000

default_ttl class-attribute instance-attribute 

default_ttl: float = 300.0

cleanup_interval class-attribute instance-attribute 

cleanup_interval: float = 60.0

enable_stats class-attribute instance-attribute 

enable_stats: bool = True

CacheEntry

Cache entry with metadata.

Source code in src/honeyhive/utils/cache.py 
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
class CacheEntry:
    """Cache entry with metadata."""

    def __init__(self, key: str, value: Any, ttl: float = 300.0):
        """Initialize cache entry.

        Args:
            key: Cache key
            value: Cached value
            ttl: Time to live in seconds
        """
        self.key = key
        self.value = value
        self.created_at = time.time()
        self.ttl = ttl
        self.access_count = 0
        self.last_accessed = self.created_at

    def is_expired(self) -> bool:
        """Check if entry is expired.

        Returns:
            True if expired, False otherwise
        """
        return time.time() - self.created_at > self.ttl

    def access(self) -> None:
        """Mark entry as accessed."""
        self.access_count += 1
        self.last_accessed = time.time()

    def get_age(self) -> float:
        """Get age of entry in seconds.

        Returns:
            Age in seconds
        """
        return time.time() - self.created_at

    def get_remaining_ttl(self) -> float:
        """Get remaining TTL in seconds.

        Returns:
            Remaining TTL in seconds
        """
        remaining = self.ttl - self.get_age()
        return max(0, remaining)

    @property
    def expiry(self) -> float:
        """Get expiry timestamp.

        Returns:
            Timestamp when entry expires
        """
        return self.created_at + self.ttl

key instance-attribute 

key = key

value instance-attribute 

value = value

created_at instance-attribute 

created_at = time()

ttl instance-attribute 

ttl = ttl

access_count instance-attribute 

access_count = 0

last_accessed instance-attribute 

last_accessed = created_at

expiry property 

expiry: float

Get expiry timestamp.

Returns:

Type Description
float

Timestamp when entry expires

is_expired 

is_expired() -> bool

Check if entry is expired.

Returns:

Type Description
bool

True if expired, False otherwise
Source code in src/honeyhive/utils/cache.py 
38
39
40
41
42
43
44
def is_expired(self) -> bool:
    """Check if entry is expired.

    Returns:
        True if expired, False otherwise
    """
    return time.time() - self.created_at > self.ttl

access 

access() -> None

Mark entry as accessed.

Source code in src/honeyhive/utils/cache.py 
46
47
48
49
def access(self) -> None:
    """Mark entry as accessed."""
    self.access_count += 1
    self.last_accessed = time.time()

get_age 

get_age() -> float

Get age of entry in seconds.

Returns:

Type Description
float

Age in seconds
Source code in src/honeyhive/utils/cache.py 
51
52
53
54
55
56
57
def get_age(self) -> float:
    """Get age of entry in seconds.

    Returns:
        Age in seconds
    """
    return time.time() - self.created_at

get_remaining_ttl 

get_remaining_ttl() -> float

Get remaining TTL in seconds.

Returns:

Type Description
float

Remaining TTL in seconds
Source code in src/honeyhive/utils/cache.py 
59
60
61
62
63
64
65
66
def get_remaining_ttl(self) -> float:
    """Get remaining TTL in seconds.

    Returns:
        Remaining TTL in seconds
    """
    remaining = self.ttl - self.get_age()
    return max(0, remaining)

Cache

In-memory cache with TTL and size limits.

Source code in src/honeyhive/utils/cache.py 
 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
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
class Cache:
    """In-memory cache with TTL and size limits."""

    def __init__(self, config: Optional[CacheConfig] = None):
        """Initialize cache.

        Args:
            config: Cache configuration
        """
        self.config = config or CacheConfig()

        # Cache storage
        self._cache: Dict[str, CacheEntry] = {}
        self._lock = threading.RLock()

        # Statistics
        self._stats = {
            "hits": 0,
            "misses": 0,
            "sets": 0,
            "deletes": 0,
            "expired": 0,
            "evictions": 0,
        }

        # Cleanup thread
        self._cleanup_thread: Optional[threading.Thread] = None
        self._stop_cleanup = threading.Event()
        self._start_cleanup_thread()

    @property
    def cache(self) -> Dict[str, CacheEntry]:
        """Get the underlying cache dictionary.

        Returns:
            Cache dictionary
        """
        return self._cache

    @property
    def hits(self) -> int:
        """Get cache hit count.

        Returns:
            Number of cache hits
        """
        return self._stats["hits"]

    @property
    def misses(self) -> int:
        """Get cache miss count.

        Returns:
            Number of cache misses
        """
        return self._stats["misses"]

    def _start_cleanup_thread(self) -> None:
        """Start cleanup thread."""
        if self.config.cleanup_interval > 0:
            self._cleanup_thread = threading.Thread(
                target=self._cleanup_worker, daemon=True
            )
            self._cleanup_thread.start()

    def _cleanup_worker(self) -> None:
        """Cleanup worker thread."""
        while not self._stop_cleanup.wait(self.config.cleanup_interval):
            self.cleanup_expired()

    def _generate_key(self, *args: Any, **kwargs: Any) -> str:
        """Generate cache key from arguments.

        Args:
            *args: Positional arguments
            **kwargs: Keyword arguments

        Returns:
            Cache key string
        """
        # Create a string representation of the arguments
        key_parts = [str(arg) for arg in args]
        key_parts.extend(f"{k}={v}" for k, v in sorted(kwargs.items()))
        key_string = "|".join(key_parts)

        # Hash the key string for consistent length
        return hashlib.md5(key_string.encode()).hexdigest()

    def generate_key(self, *args: Any, **kwargs: Any) -> str:
        """Generate cache key from arguments (public method).

        Args:
            *args: Positional arguments
            **kwargs: Keyword arguments

        Returns:
            Cache key string
        """
        return self._generate_key(*args, **kwargs)

    def get(self, key: str, default: Any = None) -> Any:
        """Get value from cache.

        Args:
            key: Cache key
            default: Default value if key not found

        Returns:
            Cached value or default
        """
        with self._lock:
            if key in self._cache:
                entry = self._cache[key]

                if entry.is_expired():
                    # Remove expired entry
                    del self._cache[key]
                    self._stats["expired"] += 1
                    self._stats["misses"] += 1
                    return default

                # Mark as accessed
                entry.access()
                self._stats["hits"] += 1
                return entry.value

            self._stats["misses"] += 1
            return default

    def set(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
        """Set value in cache.

        Args:
            key: Cache key
            value: Value to cache
            ttl: Time to live in seconds (uses default if None)
        """
        if ttl is None:
            ttl = self.config.default_ttl

        with self._lock:
            # Check if we need to evict entries
            if len(self._cache) >= self.config.max_size:
                self._evict_entries()

            # Create cache entry
            entry = CacheEntry(key, value, ttl)
            self._cache[key] = entry
            self._stats["sets"] += 1

    def delete(self, key: str) -> bool:
        """Delete key from cache.

        Args:
            key: Cache key to delete

        Returns:
            True if key was deleted, False if not found
        """
        with self._lock:
            if key in self._cache:
                del self._cache[key]
                self._stats["deletes"] += 1
                return True
            return False

    def exists(self, key: str) -> bool:
        """Check if key exists in cache.

        Args:
            key: Cache key to check

        Returns:
            True if key exists and not expired, False otherwise
        """
        with self._lock:
            if key in self._cache:
                entry = self._cache[key]
                if entry.is_expired():
                    del self._cache[key]
                    self._stats["expired"] += 1
                    return False
                return True
            return False

    def clear(self) -> None:
        """Clear all entries from cache."""
        with self._lock:
            self._cache.clear()
            self._reset_stats()

    def cleanup_expired(self) -> int:
        """Clean up expired entries.

        Returns:
            Number of entries cleaned up
        """
        cleaned = 0
        current_time = time.time()

        with self._lock:
            expired_keys = [
                key
                for key, entry in self._cache.items()
                if current_time - entry.created_at > entry.ttl
            ]

            for key in expired_keys:
                del self._cache[key]
                cleaned += 1
                self._stats["expired"] += 1

        return cleaned

    def _evict_entries(self, count: int = 1) -> None:
        """Evict entries based on LRU policy.

        Args:
            count: Number of entries to evict
        """
        if len(self._cache) < count:
            return

        # Sort entries by last accessed time (LRU)
        entries = sorted(self._cache.items(), key=lambda x: x[1].last_accessed)

        # Remove oldest entries
        for i in range(count):
            if i < len(entries):
                key, _ = entries[i]
                del self._cache[key]
                self._stats["evictions"] += 1

    def get_stats(self) -> Dict[str, Any]:
        """Get cache statistics.

        Returns:
            Dictionary with cache statistics
        """
        with self._lock:
            stats = self._stats.copy()
            stats["size"] = len(self._cache)
            stats["max_size"] = self.config.max_size
            stats["hit_rate"] = int(
                self._stats["hits"]
                / max(1, self._stats["hits"] + self._stats["misses"])
                * 100
            )
            return stats

    def _reset_stats(self) -> None:
        """Reset cache statistics."""
        for key in self._stats:
            self._stats[key] = 0

    def stats(self) -> Dict[str, Any]:
        """Get cache statistics.

        Returns:
            Dictionary with cache statistics
        """
        with self._lock:
            total_requests = self._stats["hits"] + self._stats["misses"]
            return {
                "size": len(self._cache),
                "max_size": self.config.max_size,
                "hits": self._stats["hits"],
                "misses": self._stats["misses"],
                "total_requests": total_requests,
                "hit_rate": self._stats["hits"] / max(1, total_requests),
                "sets": self._stats["sets"],
                "deletes": self._stats["deletes"],
                "expired": self._stats["expired"],
                "evictions": self._stats["evictions"],
            }

    def cleanup(self) -> None:
        """Clean up expired entries and perform maintenance."""
        self.cleanup_expired()

    def close(self) -> None:
        """Close cache and cleanup resources."""
        self._stop_cleanup.set()
        if self._cleanup_thread and self._cleanup_thread.is_alive():
            self._cleanup_thread.join(timeout=1.0)
        self.clear()

    def __enter__(self) -> "Cache":
        """Context manager entry."""
        return self

    def __exit__(
        self,
        exc_type: Optional[type],
        exc_val: Optional[BaseException],
        exc_tb: Optional[Any],
    ) -> None:
        """Context manager exit."""
        self.close()

config instance-attribute 

config = config or CacheConfig()

cache property 

cache: Dict[str, CacheEntry]

Get the underlying cache dictionary.

Returns:

Type Description
Dict[str, CacheEntry]

Cache dictionary

hits property 

hits: int

Get cache hit count.

Returns:

Type Description
int

Number of cache hits

misses property 

misses: int

Get cache miss count.

Returns:

Type Description
int

Number of cache misses

generate_key 

generate_key(*args: Any, **kwargs: Any) -> str

Generate cache key from arguments (public method).

Parameters:

Name Type Description Default
*args Any

Positional arguments

 ()
**kwargs Any

Keyword arguments

 {}

Returns:

Type Description
str

Cache key string
Source code in src/honeyhive/utils/cache.py 
166
167
168
169
170
171
172
173
174
175
176
def generate_key(self, *args: Any, **kwargs: Any) -> str:
    """Generate cache key from arguments (public method).

    Args:
        *args: Positional arguments
        **kwargs: Keyword arguments

    Returns:
        Cache key string
    """
    return self._generate_key(*args, **kwargs)

get 

get(key: str, default: Any = None) -> Any

Get value from cache.

Parameters:

Name Type Description Default
key str

Cache key

 required
default Any

Default value if key not found

 None

Returns:

Type Description
Any

Cached value or default
Source code in src/honeyhive/utils/cache.py 
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
def get(self, key: str, default: Any = None) -> Any:
    """Get value from cache.

    Args:
        key: Cache key
        default: Default value if key not found

    Returns:
        Cached value or default
    """
    with self._lock:
        if key in self._cache:
            entry = self._cache[key]

            if entry.is_expired():
                # Remove expired entry
                del self._cache[key]
                self._stats["expired"] += 1
                self._stats["misses"] += 1
                return default

            # Mark as accessed
            entry.access()
            self._stats["hits"] += 1
            return entry.value

        self._stats["misses"] += 1
        return default

set 

set(
    key: str, value: Any, ttl: Optional[float] = None
) -> None

Set value in cache.

Parameters:

Name Type Description Default
key str

Cache key

 required
value Any

Value to cache

 required
ttl Optional[float]

Time to live in seconds (uses default if None)

 None
Source code in src/honeyhive/utils/cache.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def set(self, key: str, value: Any, ttl: Optional[float] = None) -> None:
    """Set value in cache.

    Args:
        key: Cache key
        value: Value to cache
        ttl: Time to live in seconds (uses default if None)
    """
    if ttl is None:
        ttl = self.config.default_ttl

    with self._lock:
        # Check if we need to evict entries
        if len(self._cache) >= self.config.max_size:
            self._evict_entries()

        # Create cache entry
        entry = CacheEntry(key, value, ttl)
        self._cache[key] = entry
        self._stats["sets"] += 1

delete 

delete(key: str) -> bool

Delete key from cache.

Parameters:

Name Type Description Default
key str

Cache key to delete

 required

Returns:

Type Description
bool

True if key was deleted, False if not found
Source code in src/honeyhive/utils/cache.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
def delete(self, key: str) -> bool:
    """Delete key from cache.

    Args:
        key: Cache key to delete

    Returns:
        True if key was deleted, False if not found
    """
    with self._lock:
        if key in self._cache:
            del self._cache[key]
            self._stats["deletes"] += 1
            return True
        return False

exists 

exists(key: str) -> bool

Check if key exists in cache.

Parameters:

Name Type Description Default
key str

Cache key to check

 required

Returns:

Type Description
bool

True if key exists and not expired, False otherwise
Source code in src/honeyhive/utils/cache.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def exists(self, key: str) -> bool:
    """Check if key exists in cache.

    Args:
        key: Cache key to check

    Returns:
        True if key exists and not expired, False otherwise
    """
    with self._lock:
        if key in self._cache:
            entry = self._cache[key]
            if entry.is_expired():
                del self._cache[key]
                self._stats["expired"] += 1
                return False
            return True
        return False

clear 

clear() -> None

Clear all entries from cache.

Source code in src/honeyhive/utils/cache.py 
263
264
265
266
267
def clear(self) -> None:
    """Clear all entries from cache."""
    with self._lock:
        self._cache.clear()
        self._reset_stats()

cleanup_expired 

cleanup_expired() -> int

Clean up expired entries.

Returns:

Type Description
int

Number of entries cleaned up
Source code in src/honeyhive/utils/cache.py 
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def cleanup_expired(self) -> int:
    """Clean up expired entries.

    Returns:
        Number of entries cleaned up
    """
    cleaned = 0
    current_time = time.time()

    with self._lock:
        expired_keys = [
            key
            for key, entry in self._cache.items()
            if current_time - entry.created_at > entry.ttl
        ]

        for key in expired_keys:
            del self._cache[key]
            cleaned += 1
            self._stats["expired"] += 1

    return cleaned

get_stats 

get_stats() -> Dict[str, Any]

Get cache statistics.

Returns:

Type Description
Dict[str, Any]

Dictionary with cache statistics
Source code in src/honeyhive/utils/cache.py 
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
def get_stats(self) -> Dict[str, Any]:
    """Get cache statistics.

    Returns:
        Dictionary with cache statistics
    """
    with self._lock:
        stats = self._stats.copy()
        stats["size"] = len(self._cache)
        stats["max_size"] = self.config.max_size
        stats["hit_rate"] = int(
            self._stats["hits"]
            / max(1, self._stats["hits"] + self._stats["misses"])
            * 100
        )
        return stats

stats 

stats() -> Dict[str, Any]

Get cache statistics.

Returns:

Type Description
Dict[str, Any]

Dictionary with cache statistics
Source code in src/honeyhive/utils/cache.py 
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def stats(self) -> Dict[str, Any]:
    """Get cache statistics.

    Returns:
        Dictionary with cache statistics
    """
    with self._lock:
        total_requests = self._stats["hits"] + self._stats["misses"]
        return {
            "size": len(self._cache),
            "max_size": self.config.max_size,
            "hits": self._stats["hits"],
            "misses": self._stats["misses"],
            "total_requests": total_requests,
            "hit_rate": self._stats["hits"] / max(1, total_requests),
            "sets": self._stats["sets"],
            "deletes": self._stats["deletes"],
            "expired": self._stats["expired"],
            "evictions": self._stats["evictions"],
        }

cleanup 

cleanup() -> None

Clean up expired entries and perform maintenance.

Source code in src/honeyhive/utils/cache.py 
354
355
356
def cleanup(self) -> None:
    """Clean up expired entries and perform maintenance."""
    self.cleanup_expired()

close 

close() -> None

Close cache and cleanup resources.

Source code in src/honeyhive/utils/cache.py 
358
359
360
361
362
363
def close(self) -> None:
    """Close cache and cleanup resources."""
    self._stop_cleanup.set()
    if self._cleanup_thread and self._cleanup_thread.is_alive():
        self._cleanup_thread.join(timeout=1.0)
    self.clear()

FunctionCache

Function result cache decorator.

Note: too-few-public-methods disabled - Decorator classes need init/call.

Source code in src/honeyhive/utils/cache.py 
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
class FunctionCache:  # pylint: disable=too-few-public-methods
    """Function result cache decorator.

    Note: too-few-public-methods disabled - Decorator classes need __init__/__call__.
    """

    def __init__(
        self,
        cache: Optional[Cache] = None,
        ttl: Optional[float] = None,
        key_func: Optional[Callable] = None,
    ):
        """Initialize function cache.

        Args:
            cache: Cache instance to use
            ttl: Time to live for cached results
            key_func: Custom key generation function
        """
        self.cache = cache or Cache()
        self.ttl = ttl
        self.key_func = key_func

    def __call__(self, func: Callable) -> Callable:
        """Cache decorator.

        Args:
            func: Function to cache

        Returns:
            Cached function
        """

        def cached_func(*args: Any, **kwargs: Any) -> Any:
            # Generate cache key
            if self.key_func:
                key = self.key_func(func, *args, **kwargs)
            else:
                key = self.cache.generate_key(func.__name__, *args, **kwargs)

            # Try to get from cache
            result = self.cache.get(key)
            if result is not None:
                return result

            # Execute function and cache result
            result = func(*args, **kwargs)
            self.cache.set(key, result, self.ttl)
            return result

        return cached_func

cache instance-attribute 

cache = cache or Cache()

ttl instance-attribute 

ttl = ttl

key_func instance-attribute 

key_func = key_func

AsyncFunctionCache

Async function result cache decorator.

Note: too-few-public-methods disabled - Decorator classes need init/call.

Source code in src/honeyhive/utils/cache.py 
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
class AsyncFunctionCache:  # pylint: disable=too-few-public-methods
    """Async function result cache decorator.

    Note: too-few-public-methods disabled - Decorator classes need __init__/__call__.
    """

    def __init__(
        self,
        cache: Optional[Cache] = None,
        ttl: Optional[float] = None,
        key_func: Optional[Callable] = None,
    ):
        """Initialize async function cache.

        Args:
            cache: Cache instance to use
            ttl: Time to live for cached results
            key_func: Custom key generation function
        """
        self.cache = cache or Cache()
        self.ttl = ttl
        self.key_func = key_func

    def __call__(self, func: Callable) -> Callable:
        """Async cache decorator.

        Args:
            func: Async function to cache

        Returns:
            Cached async function
        """

        async def cached_func(*args: Any, **kwargs: Any) -> Any:
            # Generate cache key
            if self.key_func:
                key = self.key_func(func, *args, **kwargs)
            else:
                key = self.cache.generate_key(func.__name__, *args, **kwargs)

            # Try to get from cache
            result = self.cache.get(key)
            if result is not None:
                return result

            # Execute function and cache result
            result = await func(*args, **kwargs)
            self.cache.set(key, result, self.ttl)
            return result

        return cached_func

cache instance-attribute 

cache = cache or Cache()

ttl instance-attribute 

ttl = ttl

key_func instance-attribute 

key_func = key_func

CacheManager

Multi-instance cache manager for tracer instances.

This class provides per-instance cache management that aligns with the multi-instance tracer architecture. Each tracer instance can have its own isolated cache instances.

Source code in src/honeyhive/utils/cache.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
class CacheManager:
    """Multi-instance cache manager for tracer instances.

    This class provides per-instance cache management that aligns with
    the multi-instance tracer architecture. Each tracer instance can
    have its own isolated cache instances.
    """

    def __init__(self, instance_id: str, config: Optional[CacheConfig] = None):
        """Initialize cache manager for a specific instance.

        Args:
            instance_id: Unique identifier for the instance (e.g., tracer ID)
            config: Cache configuration
        """
        self.instance_id = instance_id
        self.config = config or CacheConfig()
        self._caches: Dict[str, Cache] = {}

    def get_cache(self, cache_name: str, config: Optional[CacheConfig] = None) -> Cache:
        """Get or create a named cache for this instance.

        Args:
            cache_name: Name of the cache (e.g., 'attributes', 'resources')
            config: Optional cache-specific configuration

        Returns:
            Cache instance for the specified name
        """
        if cache_name not in self._caches:
            cache_config = config or self.config
            self._caches[cache_name] = Cache(cache_config)

        return self._caches[cache_name]

    def close_all(self) -> None:
        """Close all caches managed by this instance."""
        for cache in self._caches.values():
            cache.close()
        self._caches.clear()

    def get_stats(self) -> Dict[str, Dict[str, Any]]:
        """Get statistics for all caches in this instance.

        Returns:
            Dictionary mapping cache names to their statistics
        """
        return {name: cache.get_stats() for name, cache in self._caches.items()}

    # Domain-specific cache methods for tracer functionality
    def get_config_value(
        self,
        config_hash: str,
        key: str,
        default: Any,
        resolver_func: Callable[[], Any],
    ) -> Any:
        """Get cached configuration value or resolve and cache it.

        Args:
            config_hash: Hash of the configuration object
            key: Configuration key
            default: Default value if not found
            resolver_func: Function to resolve the value if not cached

        Returns:
            Cached or resolved configuration value
        """
        cache = self.get_cache(
            "config",
            CacheConfig(
                max_size=100,
                default_ttl=900.0,  # 15-minute TTL for config stability
                cleanup_interval=180.0,
            ),
        )

        cache_key = f"config:{config_hash}:{key}:{hash(str(default))}"

        # Check cache first
        if cached := cache.get(cache_key):
            return cached

        # Resolve and cache
        try:
            value = resolver_func()
            cache.set(cache_key, value)
            return value
        except Exception:
            return default

    def get_cached_attributes(
        self,
        attr_key: str,
        normalizer_func: Callable[[], Any],
    ) -> Any:
        """Get cached normalized attributes or normalize and cache them.

        Args:
            attr_key: Attribute cache key
            normalizer_func: Function to normalize the attribute if not cached

        Returns:
            Cached or normalized attribute value
        """
        cache = self.get_cache(
            "attributes",
            CacheConfig(
                max_size=1000,  # High frequency operations
                default_ttl=300.0,  # 5-minute TTL
                cleanup_interval=60.0,
            ),
        )

        # Check cache first
        if cached := cache.get(attr_key):
            return cached

        # Normalize and cache
        try:
            value = normalizer_func()
            cache.set(attr_key, value)
            return value
        except Exception:
            return None

    def get_cached_resources(
        self,
        resource_key: str,
        detector_func: Callable[[], Dict[str, Any]],
    ) -> Dict[str, Any]:
        """Get cached resource detection results or detect and cache them.

        Args:
            resource_key: Resource cache key
            detector_func: Function to detect resources if not cached

        Returns:
            Cached or detected resource information
        """
        cache = self.get_cache(
            "resources",
            CacheConfig(
                max_size=50,  # Lower frequency, stable data
                default_ttl=3600.0,  # 1-hour TTL for system info
                cleanup_interval=300.0,
            ),
        )

        # Check cache first
        if cached := cache.get(resource_key):
            return cached  # type: ignore[no-any-return]

        # Detect and cache
        try:
            resources = detector_func()
            cache.set(resource_key, resources)
            return resources
        except Exception:
            return {}

instance_id instance-attribute 

instance_id = instance_id

config instance-attribute 

config = config or CacheConfig()

get_cache 

get_cache(
    cache_name: str, config: Optional[CacheConfig] = None
) -> Cache

Get or create a named cache for this instance.

Parameters:

Name Type Description Default
cache_name str

Name of the cache (e.g., 'attributes', 'resources')

 required
config Optional[CacheConfig]

Optional cache-specific configuration

 None

Returns:

Type Description
Cache

Cache instance for the specified name
Source code in src/honeyhive/utils/cache.py 
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def get_cache(self, cache_name: str, config: Optional[CacheConfig] = None) -> Cache:
    """Get or create a named cache for this instance.

    Args:
        cache_name: Name of the cache (e.g., 'attributes', 'resources')
        config: Optional cache-specific configuration

    Returns:
        Cache instance for the specified name
    """
    if cache_name not in self._caches:
        cache_config = config or self.config
        self._caches[cache_name] = Cache(cache_config)

    return self._caches[cache_name]

close_all 

close_all() -> None

Close all caches managed by this instance.

Source code in src/honeyhive/utils/cache.py 
524
525
526
527
528
def close_all(self) -> None:
    """Close all caches managed by this instance."""
    for cache in self._caches.values():
        cache.close()
    self._caches.clear()

get_stats 

get_stats() -> Dict[str, Dict[str, Any]]

Get statistics for all caches in this instance.

Returns:

Type Description
Dict[str, Dict[str, Any]]

Dictionary mapping cache names to their statistics
Source code in src/honeyhive/utils/cache.py 
530
531
532
533
534
535
536
def get_stats(self) -> Dict[str, Dict[str, Any]]:
    """Get statistics for all caches in this instance.

    Returns:
        Dictionary mapping cache names to their statistics
    """
    return {name: cache.get_stats() for name, cache in self._caches.items()}

get_config_value 

get_config_value(
    config_hash: str,
    key: str,
    default: Any,
    resolver_func: Callable[[], Any],
) -> Any

Get cached configuration value or resolve and cache it.

Parameters:

Name Type Description Default
config_hash str

Hash of the configuration object

 required
key str

Configuration key

 required
default Any

Default value if not found

 required
resolver_func Callable[[], Any]

Function to resolve the value if not cached

 required

Returns:

Type Description
Any

Cached or resolved configuration value
Source code in src/honeyhive/utils/cache.py 
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
def get_config_value(
    self,
    config_hash: str,
    key: str,
    default: Any,
    resolver_func: Callable[[], Any],
) -> Any:
    """Get cached configuration value or resolve and cache it.

    Args:
        config_hash: Hash of the configuration object
        key: Configuration key
        default: Default value if not found
        resolver_func: Function to resolve the value if not cached

    Returns:
        Cached or resolved configuration value
    """
    cache = self.get_cache(
        "config",
        CacheConfig(
            max_size=100,
            default_ttl=900.0,  # 15-minute TTL for config stability
            cleanup_interval=180.0,
        ),
    )

    cache_key = f"config:{config_hash}:{key}:{hash(str(default))}"

    # Check cache first
    if cached := cache.get(cache_key):
        return cached

    # Resolve and cache
    try:
        value = resolver_func()
        cache.set(cache_key, value)
        return value
    except Exception:
        return default

get_cached_attributes 

get_cached_attributes(
    attr_key: str, normalizer_func: Callable[[], Any]
) -> Any

Get cached normalized attributes or normalize and cache them.

Parameters:

Name Type Description Default
attr_key str

Attribute cache key

 required
normalizer_func Callable[[], Any]

Function to normalize the attribute if not cached

 required

Returns:

Type Description
Any

Cached or normalized attribute value
Source code in src/honeyhive/utils/cache.py 
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
def get_cached_attributes(
    self,
    attr_key: str,
    normalizer_func: Callable[[], Any],
) -> Any:
    """Get cached normalized attributes or normalize and cache them.

    Args:
        attr_key: Attribute cache key
        normalizer_func: Function to normalize the attribute if not cached

    Returns:
        Cached or normalized attribute value
    """
    cache = self.get_cache(
        "attributes",
        CacheConfig(
            max_size=1000,  # High frequency operations
            default_ttl=300.0,  # 5-minute TTL
            cleanup_interval=60.0,
        ),
    )

    # Check cache first
    if cached := cache.get(attr_key):
        return cached

    # Normalize and cache
    try:
        value = normalizer_func()
        cache.set(attr_key, value)
        return value
    except Exception:
        return None

get_cached_resources 

get_cached_resources(
    resource_key: str,
    detector_func: Callable[[], Dict[str, Any]],
) -> Dict[str, Any]

Get cached resource detection results or detect and cache them.

Parameters:

Name Type Description Default
resource_key str

Resource cache key

 required
detector_func Callable[[], Dict[str, Any]]

Function to detect resources if not cached

 required

Returns:

Type Description
Dict[str, Any]

Cached or detected resource information
Source code in src/honeyhive/utils/cache.py 
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
def get_cached_resources(
    self,
    resource_key: str,
    detector_func: Callable[[], Dict[str, Any]],
) -> Dict[str, Any]:
    """Get cached resource detection results or detect and cache them.

    Args:
        resource_key: Resource cache key
        detector_func: Function to detect resources if not cached

    Returns:
        Cached or detected resource information
    """
    cache = self.get_cache(
        "resources",
        CacheConfig(
            max_size=50,  # Lower frequency, stable data
            default_ttl=3600.0,  # 1-hour TTL for system info
            cleanup_interval=300.0,
        ),
    )

    # Check cache first
    if cached := cache.get(resource_key):
        return cached  # type: ignore[no-any-return]

    # Detect and cache
    try:
        resources = detector_func()
        cache.set(resource_key, resources)
        return resources
    except Exception:
        return {}

get_global_cache 

get_global_cache(
    config: Optional[CacheConfig] = None,
) -> Cache

Get or create global cache instance.

Note: This function is maintained for CLI and backward compatibility. For tracer instances, use CacheManager for proper multi-instance isolation.

Parameters:

Name Type Description Default
config Optional[CacheConfig]

Cache configuration

 None

Returns:

Type Description
Cache

Global cache instance
Source code in src/honeyhive/utils/cache.py 
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
def get_global_cache(config: Optional[CacheConfig] = None) -> Cache:
    """Get or create global cache instance.

    Note: This function is maintained for CLI and backward compatibility.
    For tracer instances, use CacheManager for proper multi-instance isolation.

    Args:
        config: Cache configuration

    Returns:
        Global cache instance
    """
    global _global_cache  # pylint: disable=global-statement

    if _global_cache is None:
        _global_cache = Cache(config)

    return _global_cache

close_global_cache 

close_global_cache() -> None

Close global cache instance.

Note: This function is maintained for CLI and backward compatibility.

Source code in src/honeyhive/utils/cache.py 
675
676
677
678
679
680
681
682
683
684
def close_global_cache() -> None:
    """Close global cache instance.

    Note: This function is maintained for CLI and backward compatibility.
    """
    global _global_cache  # pylint: disable=global-statement

    if _global_cache is not None:
        _global_cache.close()
        _global_cache = None

cache_function 

cache_function(
    ttl: Optional[float] = None,
    cache: Optional[Cache] = None,
    key_func: Optional[Callable] = None,
) -> FunctionCache

Decorator to cache function results.

Parameters:

Name Type Description Default
ttl Optional[float]

Time to live for cached results

 None
cache Optional[Cache]

Cache instance to use

 None
key_func Optional[Callable]

Custom key generation function

 None

Returns:

Type Description
FunctionCache

Function cache decorator
Source code in src/honeyhive/utils/cache.py 
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
def cache_function(
    ttl: Optional[float] = None,
    cache: Optional[Cache] = None,
    key_func: Optional[Callable] = None,
) -> FunctionCache:
    """Decorator to cache function results.

    Args:
        ttl: Time to live for cached results
        cache: Cache instance to use
        key_func: Custom key generation function

    Returns:
        Function cache decorator
    """
    return FunctionCache(cache=cache, ttl=ttl, key_func=key_func)

cache_async_function 

cache_async_function(
    ttl: Optional[float] = None,
    cache: Optional[Cache] = None,
    key_func: Optional[Callable] = None,
) -> AsyncFunctionCache

Decorator to cache async function results.

Parameters:

Name Type Description Default
ttl Optional[float]

Time to live for cached results

 None
cache Optional[Cache]

Cache instance to use

 None
key_func Optional[Callable]

Custom key generation function

 None

Returns:

Type Description
AsyncFunctionCache

Async function cache decorator
Source code in src/honeyhive/utils/cache.py 
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
def cache_async_function(
    ttl: Optional[float] = None,
    cache: Optional[Cache] = None,
    key_func: Optional[Callable] = None,
) -> AsyncFunctionCache:
    """Decorator to cache async function results.

    Args:
        ttl: Time to live for cached results
        cache: Cache instance to use
        key_func: Custom key generation function

    Returns:
        Async function cache decorator
    """
    return AsyncFunctionCache(cache=cache, ttl=ttl, key_func=key_func)