Spaces:
Running
on
Zero
Running
on
Zero
File size: 3,737 Bytes
76897aa |
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 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 |
"""Cache factory for initialising caching infrastructure.
This module provides factory functions and singleton instances for
cache management across the application.
"""
import logging
from typing import Optional
from backend.caching.redis_cache import (
FinancialDataCacheManager,
HybridCache,
)
from backend.config import settings
logger = logging.getLogger(__name__)
class CacheFactory:
"""Factory for creating and managing cache instances.
Provides singleton pattern for cache instances to ensure
consistent cache usage across the application.
"""
_cache_instance: Optional[HybridCache] = None
_manager_instance: Optional[FinancialDataCacheManager] = None
@classmethod
def get_cache(cls, force_new: bool = False) -> HybridCache:
"""Get or create HybridCache instance.
Args:
force_new: Force creation of new instance.
Returns:
HybridCache instance.
"""
if cls._cache_instance is None or force_new:
cls._cache_instance = cls.create_cache()
return cls._cache_instance
@classmethod
def get_manager(cls, force_new: bool = False) -> FinancialDataCacheManager:
"""Get or create FinancialDataCacheManager instance.
Args:
force_new: Force creation of new instance.
Returns:
FinancialDataCacheManager instance.
"""
if cls._manager_instance is None or force_new:
cache = cls.get_cache()
cls._manager_instance = FinancialDataCacheManager(cache)
return cls._manager_instance
@classmethod
def create_cache(cls) -> HybridCache:
"""Create a new HybridCache instance from settings.
Returns:
Configured HybridCache instance.
"""
if not settings.cache_enabled:
logger.info("Caching is disabled in settings")
return HybridCache(
redis_url=None,
redis_token=None,
enable_fallback=True,
)
cache = HybridCache(
redis_url=settings.upstash_redis_url,
redis_token=settings.upstash_redis_token,
enable_fallback=settings.cache_fallback_enabled,
fallback_size=settings.cache_fallback_max_size,
)
logger.info(
f"Cache initialised with backend: {cache.stats.backend.value}"
)
return cache
@classmethod
def reset(cls) -> None:
"""Reset singleton instances.
Useful for testing or when configuration changes.
"""
cls._cache_instance = None
cls._manager_instance = None
logger.info("Cache factory reset")
def get_cache() -> HybridCache:
"""Get global cache instance.
Convenience function for accessing the singleton cache.
Returns:
HybridCache instance.
Example:
>>> from backend.caching.factory import get_cache
>>> cache = get_cache()
>>> cache.set("key", "value", ttl=300)
>>> value = cache.get("key")
"""
return CacheFactory.get_cache()
def get_cache_manager() -> FinancialDataCacheManager:
"""Get global cache manager instance.
Convenience function for accessing the singleton cache manager.
Returns:
FinancialDataCacheManager instance.
Example:
>>> from backend.caching.factory import get_cache_manager
>>> manager = get_cache_manager()
>>> manager.cache_market_data("AAPL", {"price": 150.00})
>>> data = manager.get_market_data("AAPL")
"""
return CacheFactory.get_manager()
# Global instances (initialised on first import)
cache = get_cache()
cache_manager = get_cache_manager()
|