######################################################################
#
# 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)
import six
[docs]@six.add_metaclass(ABCMeta)
class AbstractCache(object):
[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):
""" 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):
""" 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):
""" 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))