Source code for b2sdk.cache

######################################################################
#
# File: b2sdk/cache.py
#
# Copyright 2019 Backblaze Inc. All Rights Reserved.
#
# License https://www.backblaze.com/using_b2_code.html
#
######################################################################

from abc import ABCMeta, abstractmethod


[docs]class AbstractCache(metaclass=ABCMeta):
[docs] def clear(self): self.set_bucket_name_cache(tuple())
[docs] @abstractmethod def get_bucket_id_or_none_from_bucket_name(self, name): pass
[docs] @abstractmethod def get_bucket_name_or_none_from_allowed(self): pass
[docs] @abstractmethod def save_bucket(self, bucket): pass
[docs] @abstractmethod def set_bucket_name_cache(self, buckets): pass
def _name_id_iterator(self, buckets): return ((bucket.name, bucket.id_) for bucket in buckets)
[docs]class DummyCache(AbstractCache): """ A cache that does nothing. """
[docs] def get_bucket_id_or_none_from_bucket_name(self, name): return None
[docs] def get_bucket_name_or_none_from_allowed(self): return None
[docs] def save_bucket(self, bucket): pass
[docs] def set_bucket_name_cache(self, buckets): pass
[docs]class InMemoryCache(AbstractCache): """ A cache that stores the information in memory. """
[docs] def __init__(self): self.name_id_map = {} self.bucket_name = ''
[docs] def get_bucket_id_or_none_from_bucket_name(self, name): return self.name_id_map.get(name)
[docs] def get_bucket_name_or_none_from_allowed(self): return self.bucket_name
[docs] def save_bucket(self, bucket): self.name_id_map[bucket.name] = bucket.id_
[docs] def set_bucket_name_cache(self, buckets): self.name_id_map = dict(self._name_id_iterator(buckets))
[docs]class AuthInfoCache(AbstractCache): """ A cache that stores data persistently in StoredAccountInfo. """
[docs] def __init__(self, info): self.info = info
[docs] def get_bucket_id_or_none_from_bucket_name(self, name): return self.info.get_bucket_id_or_none_from_bucket_name(name)
[docs] def get_bucket_name_or_none_from_allowed(self): return self.info.get_bucket_name_or_none_from_allowed()
[docs] def save_bucket(self, bucket): self.info.save_bucket(bucket)
[docs] def set_bucket_name_cache(self, buckets): self.info.refresh_entire_bucket_name_cache(self._name_id_iterator(buckets))