python类cache()的实例源码

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def enable(self):
        # Just patch default cache to always find nothing
        # since drf-extensions stores the instance of the cache everywhere,
        # which makes it impossible to replace with DummyCache
        def mocked_cache_get(key, default=None, version=None):
            return default
        self.patch = patch.object(cache_module.cache, 'get', mocked_cache_get)
        self.patch.start()

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def __init__(self, lock_key, timeout=3, cache=None, nowait=False):
        if cache is None:
            self.cache = default_cache
        else:
            self.cache = cache
        self.timeout = timeout
        self.lock_key = lock_key
        self.nowait = nowait

        self.__acquired_at = None

def acquire(self):
        """
        Attempt to acquire the lock, returning a boolean that represents if the
        lock is held.
        """
        # NOTE: This isn't API compatible with the standard Python
        # ``Lock.acquire`` method signature. It may make sense to make these
        # compatible in the future, but that would also require changes to the
        # the constructor: https://docs.python.org/2/library/threading.html#lock-objects

        time_remaining = self.seconds_remaining
        if time_remaining:
            raise LockAlreadyHeld('Tried to acquire lock that is already held, %.3fs remaining: %r' % (time_remaining, self))

        self.__acquired_at = None

        delay = 0.01 + random.random() / 10
        for i in xrange(int(self.timeout // delay)):
            if i != 0:
                sleep(delay)

            attempt_started_at = time()
            if self.cache.add(self.lock_key, '', self.timeout):
                self.__acquired_at = attempt_started_at
                break

            if self.nowait:
                break

        return self.__acquired_at is not None

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)

def get_cached_choices(self):
        if not self.cache_config['enabled']:
            return None
        c = caches(self.cache_config['cache'])
        return c.get(self.cache_config['key']%self.field_path)

def set_cached_choices(self,choices):
        if not self.cache_config['enabled']:
            return
        c = caches(self.cache_config['cache'])
        return c.set(self.cache_config['key']%self.field_path,choices)