python redis cache decorator

redis_cache.py

import json
import logging
from functools import wraps, _make_key
from inspect import getfullargspec

import redis

logger = logging.getLogger(__name__)
rs = redis.StrictRedis()


def fullname(o):
    """
    return full doted path of python object.
    >>> import arrow
    >>> now = arrow.now()
    >>> fullname(now)
    'arrow.arrow.Arrow'
    >>> fullname(arrow.now)
    'arrow.api.now'
    >>> fullname(now.shift)
    'arrow.arrow.Arrow.shift'
    >>> fullname(arrow.Arrow.now)
    'arrow.arrow.Arrow.now'
    """
    try:
        module = o.__module__
    except AttributeError:
        module = o.__class__.__module__
    try:
        name = o.__qualname__
    except AttributeError:
        name = o.__class__.__name__
    if module == 'builtins' or module is None:
        return name
    return f'{module}.{name}'


def redis_cache(expire=None):
    def decorator(func):
        key = fullname(func)
        fas = getfullargspec(func)
        has_args = bool(fas.args or fas.varargs
                        or fas.varkw or fas.kwonlyargs)
        if not expire and has_args:
            raise ValueError('Expire second is required when decorated function has any args.')

        def get_key(*args, **kwargs):
            if has_args:
                args_hash = hash(_make_key(args, kwargs, False))
                full_key = f'_cache:{key}{args_hash}'
            else:
                full_key = f'_cache:{key}'
            return full_key

        @wraps(func)
        def wrapper(*args, **kwargs):
            full_key = get_key(*args, **kwargs)
            cached = rs.get(full_key)
            if cached:
                result = json.loads(cached)
            else:
                result = func(*args, **kwargs)
                rs.set(full_key, json.dumps(result), ex=expire)
            logger.debug('Calling function %s with args %s and kwargs %s, '
                         f'{"shooting" if cached else "missing"} redis cache with key %s.',
                         repr(key), repr(args), repr(kwargs), repr(full_key)
                         )
            return result

        def cache(*args, **kwargs):
            full_key = get_key(*args, **kwargs)
            cached = rs.get(full_key)
            if cached:
                result = json.loads(cached)
            else:
                result = None
            logger.debug('Calling function %s with args %s and kwargs %s, '
                         f'{"shooting" if cached else "missing"} redis cache with key %s.',
                         repr(key), repr(args), repr(kwargs), repr(full_key)
                         )
            return result

        def clear(*args, **kwargs):
            """Clear cache."""
            full_key = get_key(*args, **kwargs)
            rs.delete(full_key)

        wrapper.get_key = get_key
        wrapper.cache = cache
        wrapper.clear = clear
        wrapper.func = func
        return wrapper
    return decorator


# Usage example, notice that return value of decorated function must json serializable.
# Change serializer function as you needed.
import random


@redis_cache(10)
def seed():
    return random.random()


# Call function.
seed()
# Get redis key.
seed.make_key()
# Get result from cache or None.
seed.cache()
# Clear cache.
seed.clear()
# Bypass cache, call raw function.
seed.func()

# Notice: when used to decorate class method or instance method,
# key will be difference between processes or instances.
# Because default __hash__ method will take memory address into account.
# If you want to generate consistent result, you should rewrite __hash__ method.
# For classmethod, use meta class.
class HashMeta(type):
    def __new__(mcs, name, bases, attrs):
        attrs['__hash__'] = mcs.__hash__
        return super().__new__(mcs, name, bases, attrs)

    def __hash__(cls):
        return 0


class A(metaclass=HashMeta):
    @classmethod
    @redis_cache(expire=24 * 3600)
    def a(cls):
        pass

    @redis_cache(expire=24 * 3600)
    def b(self):
        pass