# python/01-multi-level-caching.py"""Multi-level caching architecture and strategies"""fromtypingimportList,Dict,Any,Optional,Union,Tuple,Callablefromdataclassesimportdataclass,asdictfromenumimportEnumimporttimeimportjsonimporthashlibimportpickleimportthreadingfromdatetimeimportdatetime,timedeltaimportloggingfromcollectionsimportOrderedDictimportredisfromredisimportConnectionPoolasRedisConnectionPoolimportmemcachedfromfunctoolsimportwrapsimportasyncioimportaioredisfromcontextlibimportasynccontextmanager# Configure logginglogging.basicConfig(level=logging.INFO)logger=logging.getLogger(__name__)classCacheLevel(Enum):"""Cache level enumeration"""L1="l1"# In-memory cacheL2="l2"# Redis cacheL3="l3"# Database cacheclassCacheStrategy(Enum):"""Cache strategy enumeration"""LRU="lru"LFU="lfu"TTL="ttl"WRITE_THROUGH="write_through"WRITE_BACK="write_back"@dataclassclassCacheEntry:"""Cache entry definition"""key:strvalue:Anycreated_at:datetimeaccessed_at:datetimeaccess_count:int=0ttl:Optional[int]=Nonesize:int=0defis_expired(self)->bool:"""Check if cache entry is expired"""ifself.ttlisNone:returnFalsereturndatetime.utcnow()>self.created_at+timedelta(seconds=self.ttl)defupdate_access(self)->None:"""Update access information"""self.accessed_at=datetime.utcnow()self.access_count+=1classL1Cache:"""L1 In-memory cache implementation"""def__init__(self,max_size:int=1000,strategy:CacheStrategy=CacheStrategy.LRU):self.max_size=max_sizeself.strategy=strategyself.cache:OrderedDict[str,CacheEntry]=OrderedDict()self.lock=threading.Lock()self.metrics={"hits":0,"misses":0,"evictions":0,"size":0}defget(self,key:str)->Optional[Any]:"""Get value from L1 cache"""withself.lock:ifkeyinself.cache:entry=self.cache[key]ifentry.is_expired():delself.cache[key]self.metrics["misses"]+=1returnNone# Update access informationentry.update_access()# Move to end for LRUifself.strategy==CacheStrategy.LRU:self.cache.move_to_end(key)self.metrics["hits"]+=1returnentry.valueself.metrics["misses"]+=1returnNonedefset(self,key:str,value:Any,ttl:Optional[int]=None)->None:"""Set value in L1 cache"""withself.lock:# Calculate sizesize=len(pickle.dumps(value))entry=CacheEntry(key=key,value=value,created_at=datetime.utcnow(),accessed_at=datetime.utcnow(),ttl=ttl,size=size)# Remove existing entry if presentifkeyinself.cache:delself.cache[key]# Add new entryself.cache[key]=entry# Evict if over capacitywhilelen(self.cache)>self.max_size:self._evict_entry()defdelete(self,key:str)->bool:"""Delete key from L1 cache"""withself.lock:ifkeyinself.cache:delself.cache[key]returnTruereturnFalsedefclear(self)->None:"""Clear L1 cache"""withself.lock:self.cache.clear()def_evict_entry(self)->None:"""Evict entry based on strategy"""ifnotself.cache:returnifself.strategy==CacheStrategy.LRU:# Remove least recently usedself.cache.popitem(last=False)elifself.strategy==CacheStrategy.LFU:# Remove least frequently usedlfu_key=min(self.cache.keys(),key=lambdak:self.cache[k].access_count)delself.cache[lfu_key]self.metrics["evictions"]+=1defget_metrics(self)->Dict[str,Any]:"""Get cache metrics"""total_requests=self.metrics["hits"]+self.metrics["misses"]hit_rate=(self.metrics["hits"]/total_requests*100)iftotal_requests>0else0return{**self.metrics,"hit_rate":hit_rate,"current_size":len(self.cache),"max_size":self.max_size}classL2Cache:"""L2 Redis cache implementation"""def__init__(self,redis_pool:RedisConnectionPool,key_prefix:str="cache:"):self.redis_pool=redis_poolself.key_prefix=key_prefixself.client=redis.Redis(connection_pool=redis_pool)self.metrics={"hits":0,"misses":0,"errors":0}defget(self,key:str)->Optional[Any]:"""Get value from L2 cache"""try:full_key=f"{self.key_prefix}{key}"value=self.client.get(full_key)ifvalueisnotNone:self.metrics["hits"]+=1returnpickle.loads(value)else:self.metrics["misses"]+=1returnNoneexceptExceptionase:self.metrics["errors"]+=1logger.error(f"L2 cache get error: {e}")returnNonedefset(self,key:str,value:Any,ttl:Optional[int]=None)->bool:"""Set value in L2 cache"""try:full_key=f"{self.key_prefix}{key}"serialized_value=pickle.dumps(value)ifttl:self.client.setex(full_key,ttl,serialized_value)else:self.client.set(full_key,serialized_value)returnTrueexceptExceptionase:self.metrics["errors"]+=1logger.error(f"L2 cache set error: {e}")returnFalsedefdelete(self,key:str)->bool:"""Delete key from L2 cache"""try:full_key=f"{self.key_prefix}{key}"result=self.client.delete(full_key)returnresult>0exceptExceptionase:self.metrics["errors"]+=1logger.error(f"L2 cache delete error: {e}")returnFalsedefget_metrics(self)->Dict[str,Any]:"""Get cache metrics"""total_requests=self.metrics["hits"]+self.metrics["misses"]hit_rate=(self.metrics["hits"]/total_requests*100)iftotal_requests>0else0return{**self.metrics,"hit_rate":hit_rate}classAsyncL2Cache:"""Async L2 Redis cache implementation"""def__init__(self,redis_pool,key_prefix:str="cache:"):self.redis_pool=redis_poolself.key_prefix=key_prefixself.metrics={"hits":0,"misses":0,"errors":0}asyncdefget(self,key:str)->Optional[Any]:"""Get value from async L2 cache"""try:full_key=f"{self.key_prefix}{key}"value=awaitself.redis_pool.get(full_key)ifvalueisnotNone:self.metrics["hits"]+=1returnpickle.loads(value)else:self.metrics["misses"]+=1returnNoneexceptExceptionase:self.metrics["errors"]+=1logger.error(f"Async L2 cache get error: {e}")returnNoneasyncdefset(self,key:str,value:Any,ttl:Optional[int]=None)->bool:"""Set value in async L2 cache"""try:full_key=f"{self.key_prefix}{key}"serialized_value=pickle.dumps(value)ifttl:awaitself.redis_pool.setex(full_key,ttl,serialized_value)else:awaitself.redis_pool.set(full_key,serialized_value)returnTrueexceptExceptionase:self.metrics["errors"]+=1logger.error(f"Async L2 cache set error: {e}")returnFalseasyncdefdelete(self,key:str)->bool:"""Delete key from async L2 cache"""try:full_key=f"{self.key_prefix}{key}"result=awaitself.redis_pool.delete(full_key)returnresult>0exceptExceptionase:self.metrics["errors"]+=1logger.error(f"Async L2 cache delete error: {e}")returnFalseclassMultiLevelCache:"""Multi-level cache implementation"""def__init__(self,l1_cache:L1Cache,l2_cache:L2Cache):self.l1_cache=l1_cacheself.l2_cache=l2_cacheself.metrics={"l1_hits":0,"l2_hits":0,"misses":0,"writes":0}defget(self,key:str)->Optional[Any]:"""Get value from multi-level cache"""# Try L1 cache firstvalue=self.l1_cache.get(key)ifvalueisnotNone:self.metrics["l1_hits"]+=1returnvalue# Try L2 cachevalue=self.l2_cache.get(key)ifvalueisnotNone:self.metrics["l2_hits"]+=1# Populate L1 cacheself.l1_cache.set(key,value)returnvalueself.metrics["misses"]+=1returnNonedefset(self,key:str,value:Any,ttl:Optional[int]=None)->None:"""Set value in multi-level cache"""# Set in both levelsself.l1_cache.set(key,value,ttl)self.l2_cache.set(key,value,ttl)self.metrics["writes"]+=1defdelete(self,key:str)->None:"""Delete key from multi-level cache"""self.l1_cache.delete(key)self.l2_cache.delete(key)defget_metrics(self)->Dict[str,Any]:"""Get multi-level cache metrics"""total_requests=self.metrics["l1_hits"]+self.metrics["l2_hits"]+self.metrics["misses"]l1_hit_rate=(self.metrics["l1_hits"]/total_requests*100)iftotal_requests>0else0l2_hit_rate=(self.metrics["l2_hits"]/total_requests*100)iftotal_requests>0else0overall_hit_rate=((self.metrics["l1_hits"]+self.metrics["l2_hits"])/total_requests*100)iftotal_requests>0else0return{**self.metrics,"l1_hit_rate":l1_hit_rate,"l2_hit_rate":l2_hit_rate,"overall_hit_rate":overall_hit_rate,"l1_metrics":self.l1_cache.get_metrics(),"l2_metrics":self.l2_cache.get_metrics()}classCacheDecorator:"""Cache decorator for function results"""def__init__(self,cache:MultiLevelCache,ttl:Optional[int]=None,key_prefix:str=""):self.cache=cacheself.ttl=ttlself.key_prefix=key_prefixdef__call__(self,func:Callable)->Callable:@wraps(func)defwrapper(*args,**kwargs):# Generate cache keycache_key=self._generate_cache_key(func.__name__,args,kwargs)# Try to get from cachecached_result=self.cache.get(cache_key)ifcached_resultisnotNone:returncached_result# Execute functionresult=func(*args,**kwargs)# Cache resultself.cache.set(cache_key,result,self.ttl)returnresultreturnwrapperdef_generate_cache_key(self,func_name:str,args:tuple,kwargs:dict)->str:"""Generate cache key from function arguments"""key_data={"func_name":func_name,"args":args,"kwargs":sorted(kwargs.items())}key_string=json.dumps(key_data,sort_keys=True)returnf"{self.key_prefix}{hashlib.md5(key_string.encode()).hexdigest()}"classCacheInvalidation:"""Cache invalidation strategies"""def__init__(self,cache:MultiLevelCache):self.cache=cacheself.invalidation_patterns={}defadd_pattern(self,pattern:str,keys:List[str])->None:"""Add invalidation pattern"""self.invalidation_patterns[pattern]=keysdefinvalidate_by_pattern(self,pattern:str)->int:"""Invalidate keys matching pattern"""ifpatternnotinself.invalidation_patterns:return0invalidated_count=0forkeyinself.invalidation_patterns[pattern]:ifself.cache.l1_cache.delete(key):invalidated_count+=1ifself.cache.l2_cache.delete(key):invalidated_count+=1returninvalidated_countdefinvalidate_by_prefix(self,prefix:str)->int:"""Invalidate keys with prefix"""invalidated_count=0# Invalidate L1 cachewithself.cache.l1_cache.lock:keys_to_delete=[keyforkeyinself.cache.l1_cache.cache.keys()ifkey.startswith(prefix)]forkeyinkeys_to_delete:ifself.cache.l1_cache.delete(key):invalidated_count+=1# Invalidate L2 cachetry:# This would use Redis SCAN in real implementationpassexceptExceptionase:logger.error(f"Error invalidating L2 cache by prefix: {e}")returninvalidated_count# Usage examplesdefexample_multi_level_caching():"""Example multi-level caching usage"""# Create L1 cachel1_cache=L1Cache(max_size=100,strategy=CacheStrategy.LRU)# Create Redis connection poolredis_pool=RedisConnectionPool(host='localhost',port=6379,db=0)# Create L2 cachel2_cache=L2Cache(redis_pool,key_prefix="app:")# Create multi-level cachemulti_cache=MultiLevelCache(l1_cache,l2_cache)# Set valuesmulti_cache.set("user:1",{"id":1,"name":"John"},ttl=3600)multi_cache.set("user:2",{"id":2,"name":"Jane"},ttl=3600)# Get valuesuser1=multi_cache.get("user:1")print(f"Retrieved user1: {user1}")# Get cache metricsmetrics=multi_cache.get_metrics()print(f"Cache metrics: {metrics}")# Cache decoratorcache_decorator=CacheDecorator(multi_cache,ttl=300)@cache_decoratordefexpensive_calculation(n:int)->int:time.sleep(0.1)# Simulate expensive operationreturnn*n# Test cached functionresult1=expensive_calculation(10)result2=expensive_calculation(10)# Should be cachedprint(f"Calculation results: {result1}, {result2}")# Cache invalidationinvalidation=CacheInvalidation(multi_cache)invalidation.add_pattern("users",["user:1","user:2"])invalidated=invalidation.invalidate_by_pattern("users")print(f"Invalidated {invalidated} cache entries")
# python/02-distributed-caching.py"""Distributed caching patterns and consistency strategies"""fromtypingimportList,Dict,Any,Optional,Union,Tupleimporttimeimportjsonimporthashlibimportasynciofromdatetimeimportdatetime,timedeltaimportloggingfromdataclassesimportdataclassfromenumimportEnumimportredisfromredisimportConnectionPoolasRedisConnectionPoolimportaioredisfromconsistent_hashimportConsistentHashimportthreadingfromconcurrent.futuresimportThreadPoolExecutorlogger=logging.getLogger(__name__)classConsistencyLevel(Enum):"""Consistency level enumeration"""EVENTUAL="eventual"STRONG="strong"WEAK="weak"classDistributedCache:"""Distributed cache implementation"""def__init__(self,nodes:List[str],consistency_level:ConsistencyLevel=ConsistencyLevel.EVENTUAL):self.nodes=nodesself.consistency_level=consistency_levelself.consistent_hash=ConsistentHash(nodes)self.redis_pools={}self.initialize_connections()self.metrics={"hits":0,"misses":0,"errors":0,"replications":0}definitialize_connections(self)->None:"""Initialize connections to all nodes"""fornodeinself.nodes:try:pool=RedisConnectionPool.from_url(f"redis://{node}")self.redis_pools[node]=redis.Redis(connection_pool=pool)exceptExceptionase:logger.error(f"Failed to connect to node {node}: {e}")defget_node(self,key:str)->str:"""Get node for key using consistent hashing"""returnself.consistent_hash.get_node(key)defget(self,key:str)->Optional[Any]:"""Get value from distributed cache"""try:# Get primary nodeprimary_node=self.get_node(key)# Try primary node firstvalue=self._get_from_node(primary_node,key)ifvalueisnotNone:self.metrics["hits"]+=1returnvalue# Try other nodes for eventual consistencyifself.consistency_level==ConsistencyLevel.EVENTUAL:fornodeinself.nodes:ifnode!=primary_node:value=self._get_from_node(node,key)ifvalueisnotNone:# Update primary nodeself._set_in_node(primary_node,key,value)self.metrics["hits"]+=1returnvalueself.metrics["misses"]+=1returnNoneexceptExceptionase:self.metrics["errors"]+=1logger.error(f"Distributed cache get error: {e}")returnNonedefset(self,key:str,value:Any,ttl:Optional[int]=None)->bool:"""Set value in distributed cache"""try:# Get primary nodeprimary_node=self.get_node(key)# Set in primary nodesuccess=self._set_in_node(primary_node,key,value,ttl)ifnotsuccess:returnFalse# Replicate to other nodes based on consistency levelifself.consistency_level==ConsistencyLevel.STRONG:# Synchronous replicationfornodeinself.nodes:ifnode!=primary_node:self._set_in_node(node,key,value,ttl)self.metrics["replications"]+=1elifself.consistency_level==ConsistencyLevel.EVENTUAL:# Asynchronous replicationself._async_replicate(key,value,ttl,primary_node)returnTrueexceptExceptionase:self.metrics["errors"]+=1logger.error(f"Distributed cache set error: {e}")returnFalsedefdelete(self,key:str)->bool:"""Delete key from distributed cache"""try:# Get primary nodeprimary_node=self.get_node(key)# Delete from primary nodesuccess=self._delete_from_node(primary_node,key)# Delete from other nodesfornodeinself.nodes:ifnode!=primary_node:self._delete_from_node(node,key)returnsuccessexceptExceptionase:self.metrics["errors"]+=1logger.error(f"Distributed cache delete error: {e}")returnFalsedef_get_from_node(self,node:str,key:str)->Optional[Any]:"""Get value from specific node"""try:ifnodeinself.redis_pools:value=self.redis_pools[node].get(key)returnjson.loads(value)ifvalueelseNoneexceptExceptionase:logger.error(f"Error getting from node {node}: {e}")returnNonedef_set_in_node(self,node:str,key:str,value:Any,ttl:Optional[int]=None)->bool:"""Set value in specific node"""try:ifnodeinself.redis_pools:serialized_value=json.dumps(value)ifttl:self.redis_pools[node].setex(key,ttl,serialized_value)else:self.redis_pools[node].set(key,serialized_value)returnTrueexceptExceptionase:logger.error(f"Error setting in node {node}: {e}")returnFalsedef_delete_from_node(self,node:str,key:str)->bool:"""Delete key from specific node"""try:ifnodeinself.redis_pools:result=self.redis_pools[node].delete(key)returnresult>0exceptExceptionase:logger.error(f"Error deleting from node {node}: {e}")returnFalsedef_async_replicate(self,key:str,value:Any,ttl:Optional[int],exclude_node:str)->None:"""Asynchronously replicate to other nodes"""defreplicate():fornodeinself.nodes:ifnode!=exclude_node:self._set_in_node(node,key,value,ttl)self.metrics["replications"]+=1# Run in background threadthread=threading.Thread(target=replicate)thread.daemon=Truethread.start()defget_metrics(self)->Dict[str,Any]:"""Get distributed cache metrics"""total_requests=self.metrics["hits"]+self.metrics["misses"]hit_rate=(self.metrics["hits"]/total_requests*100)iftotal_requests>0else0return{**self.metrics,"hit_rate":hit_rate,"nodes":len(self.nodes),"consistency_level":self.consistency_level.value}classCacheSharding:"""Cache sharding implementation"""def__init__(self,shards:List[str],shard_key_func:Optional[Callable]=None):self.shards=shardsself.shard_key_func=shard_key_funcorself._default_shard_keyself.shard_connections={}self.initialize_shards()definitialize_shards(self)->None:"""Initialize connections to all shards"""forshardinself.shards:try:pool=RedisConnectionPool.from_url(f"redis://{shard}")self.shard_connections[shard]=redis.Redis(connection_pool=pool)exceptExceptionase:logger.error(f"Failed to connect to shard {shard}: {e}")def_default_shard_key(self,key:str)->str:"""Default shard key function"""hash_value=int(hashlib.md5(key.encode()).hexdigest(),16)returnself.shards[hash_value%len(self.shards)]defget_shard(self,key:str)->str:"""Get shard for key"""returnself.shard_key_func(key)defget(self,key:str)->Optional[Any]:"""Get value from sharded cache"""shard=self.get_shard(key)try:ifshardinself.shard_connections:value=self.shard_connections[shard].get(key)returnjson.loads(value)ifvalueelseNoneexceptExceptionase:logger.error(f"Error getting from shard {shard}: {e}")returnNonedefset(self,key:str,value:Any,ttl:Optional[int]=None)->bool:"""Set value in sharded cache"""shard=self.get_shard(key)try:ifshardinself.shard_connections:serialized_value=json.dumps(value)ifttl:self.shard_connections[shard].setex(key,ttl,serialized_value)else:self.shard_connections[shard].set(key,serialized_value)returnTrueexceptExceptionase:logger.error(f"Error setting in shard {shard}: {e}")returnFalsedefdelete(self,key:str)->bool:"""Delete key from sharded cache"""shard=self.get_shard(key)try:ifshardinself.shard_connections:result=self.shard_connections[shard].delete(key)returnresult>0exceptExceptionase:logger.error(f"Error deleting from shard {shard}: {e}")returnFalseclassCacheWarming:"""Cache warming strategies"""def__init__(self,cache:Union[MultiLevelCache,DistributedCache]):self.cache=cacheself.warming_tasks=[]self.warming_metrics={"warmed_keys":0,"failed_warms":0,"warming_time":0.0}defadd_warming_task(self,key:str,value_func:Callable,ttl:Optional[int]=None)->None:"""Add cache warming task"""self.warming_tasks.append({"key":key,"value_func":value_func,"ttl":ttl})defwarm_cache(self)->Dict[str,Any]:"""Warm cache with all tasks"""start_time=time.time()warmed_keys=0failed_warms=0fortaskinself.warming_tasks:try:# Execute value functionvalue=task["value_func"]()# Set in cacheself.cache.set(task["key"],value,task["ttl"])warmed_keys+=1exceptExceptionase:logger.error(f"Failed to warm cache for key {task['key']}: {e}")failed_warms+=1warming_time=time.time()-start_timeself.warming_metrics["warmed_keys"]+=warmed_keysself.warming_metrics["failed_warms"]+=failed_warmsself.warming_metrics["warming_time"]+=warming_timereturn{"warmed_keys":warmed_keys,"failed_warms":failed_warms,"warming_time":warming_time}defget_warming_metrics(self)->Dict[str,Any]:"""Get cache warming metrics"""returnself.warming_metrics.copy()classCacheMonitoring:"""Cache monitoring and alerting"""def__init__(self,cache:Union[MultiLevelCache,DistributedCache]):self.cache=cacheself.monitoring_metrics={}self.alerts=[]defcheck_cache_health(self)->Dict[str,Any]:"""Check cache health"""health_status={"healthy":True,"issues":[],"timestamp":datetime.utcnow().isoformat()}# Check hit ratemetrics=self.cache.get_metrics()hit_rate=metrics.get("overall_hit_rate",0)ifhit_rate<50:# Low hit rate thresholdhealth_status["healthy"]=Falsehealth_status["issues"].append(f"Low hit rate: {hit_rate:.2f}%")# Check error rateif"errors"inmetrics:error_rate=metrics["errors"]/max(metrics.get("hits",0)+metrics.get("misses",0),1)*100iferror_rate>5:# High error rate thresholdhealth_status["healthy"]=Falsehealth_status["issues"].append(f"High error rate: {error_rate:.2f}%")returnhealth_statusdefadd_alert(self,condition:Callable,message:str)->None:"""Add monitoring alert"""self.alerts.append({"condition":condition,"message":message})defcheck_alerts(self)->List[str]:"""Check all alerts"""triggered_alerts=[]foralertinself.alerts:try:ifalert["condition"]():triggered_alerts.append(alert["message"])exceptExceptionase:logger.error(f"Error checking alert: {e}")returntriggered_alerts# Usage examplesdefexample_distributed_caching():"""Example distributed caching usage"""# Create distributed cachenodes=["redis-node-1:6379","redis-node-2:6379","redis-node-3:6379"]distributed_cache=DistributedCache(nodes,ConsistencyLevel.EVENTUAL)# Set valuesdistributed_cache.set("user:1",{"id":1,"name":"John"},ttl=3600)distributed_cache.set("user:2",{"id":2,"name":"Jane"},ttl=3600)# Get valuesuser1=distributed_cache.get("user:1")print(f"Retrieved user1: {user1}")# Get metricsmetrics=distributed_cache.get_metrics()print(f"Distributed cache metrics: {metrics}")# Cache shardingshards=["redis-shard-1:6379","redis-shard-2:6379","redis-shard-3:6379"]sharded_cache=CacheSharding(shards)# Use sharded cachesharded_cache.set("product:1",{"id":1,"name":"Laptop"},ttl=1800)product=sharded_cache.get("product:1")print(f"Retrieved product: {product}")# Cache warmingwarming=CacheWarming(distributed_cache)# Add warming taskswarming.add_warming_task("popular_users",lambda:[{"id":i,"name":f"User{i}"}foriinrange(1,11)])warming.add_warming_task("config",lambda:{"theme":"dark","language":"en"})# Warm cachewarming_result=warming.warm_cache()print(f"Cache warming result: {warming_result}")# Cache monitoringmonitoring=CacheMonitoring(distributed_cache)# Add alertsmonitoring.add_alert(lambda:distributed_cache.get_metrics().get("overall_hit_rate",0)<50,"Low cache hit rate detected")# Check healthhealth=monitoring.check_cache_health()print(f"Cache health: {health}")# Check alertsalerts=monitoring.check_alerts()ifalerts:print(f"Triggered alerts: {alerts}")