How to use the cachecontrol.cache.BaseCache function in CacheControl

To help you get started, we’ve selected a few CacheControl examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github ionrock / cachecontrol / cachecontrol / caches / file_cache.py View on Github external
# Open our file, the use of os.O_CREAT | os.O_EXCL will ensure that if a
    # race condition happens between the os.remove and this line, that an
    # error will be raised. Because we utilize a lockfile this should only
    # happen if someone is attempting to attack us.
    fd = os.open(filename, flags, fmode)
    try:
        return os.fdopen(fd, "wb")

    except:
        # An error occurred wrapping our FD in a file object
        os.close(fd)
        raise


class FileCache(BaseCache):

    def __init__(
        self,
        directory,
        forever=False,
        filemode=0o0600,
        dirmode=0o0700,
        use_dir_lock=None,
        lock_class=None,
    ):

        if use_dir_lock is not None and lock_class is not None:
            raise ValueError("Cannot use use_dir_lock and lock_class together")

        try:
            from lockfile import LockFile
github ionrock / cachecontrol / cachecontrol / caches / redis_cache.py View on Github external
from __future__ import division

from datetime import datetime
from cachecontrol.cache import BaseCache


class RedisCache(BaseCache):

    def __init__(self, conn):
        self.conn = conn

    def get(self, key):
        return self.conn.get(key)

    def set(self, key, value, expires=None):
        if not expires:
            self.conn.set(key, value)
        else:
            expires = expires - datetime.utcnow()
            self.conn.setex(key, int(expires.total_seconds()), value)

    def delete(self, key):
        self.conn.delete(key)
github ionrock / cachecontrol / cachecontrol / cache.py View on Github external
class BaseCache(object):

    def get(self, key):
        raise NotImplementedError()

    def set(self, key, value):
        raise NotImplementedError()

    def delete(self, key):
        raise NotImplementedError()

    def close(self):
        pass


class DictCache(BaseCache):

    def __init__(self, init_dict=None):
        self.lock = Lock()
        self.data = init_dict or {}

    def get(self, key):
        return self.data.get(key, None)

    def set(self, key, value):
        with self.lock:
            self.data.update({key: value})

    def delete(self, key):
        with self.lock:
            if key in self.data:
                self.data.pop(key)
github ionrock / cachecontrol / cachecontrol / cache.py View on Github external
from cPickle import loads, dumps, HIGHEST_PROTOCOL


class BaseCache(object):

    def get(self, key):
        raise NotImplemented()

    def set(self, key, value):
        raise NotImplemented()

    def delete(self, key):
        raise NotImplemented()


class DictCache(BaseCache):

    def __init__(self, init_dict=None):
        self.lock = Lock()
        self.data = init_dict or {}

    def get(self, key):
        data = self.data.get(key, None)
        if data is not None:
            try:
                data = loads(data)
            except ValueError:
                data = None
        return data

    def set(self, key, value):
        with self.lock:
github pymedusa / Medusa / ext / cachecontrol / caches / redis_cache.py View on Github external
from datetime import datetime
from cachecontrol.cache import BaseCache


def total_seconds(td):
    """Python 2.6 compatability"""
    if hasattr(td, 'total_seconds'):
        return int(td.total_seconds())

    ms = td.microseconds
    secs = (td.seconds + td.days * 24 * 3600)
    return int((ms + secs * 10**6) / 10**6)


class RedisCache(BaseCache):

    def __init__(self, conn):
        self.conn = conn

    def get(self, key):
        return self.conn.get(key)

    def set(self, key, value, expires=None):
        if not expires:
            self.conn.set(key, value)
        else:
            expires = expires - datetime.utcnow()
            self.conn.setex(key, total_seconds(expires), value)

    def delete(self, key):
        self.conn.delete(key)