Source code for aiocache.factory

from copy import deepcopy


def _class_from_string(class_path):
    class_name = class_path.split('.')[-1]
    module_name = class_path.rstrip(class_name).rstrip(".")
    return getattr(__import__(module_name, fromlist=[class_name]), class_name)


def _create_cache(cache, serializer=None, plugins=None, **kwargs):

    if serializer is not None:
        cls = serializer.pop("class")
        cls = _class_from_string(cls) if isinstance(cls, str) else cls
        serializer = cls(**serializer)

    plugins_instances = []
    if plugins is not None:
        for plugin in plugins:
            cls = plugin.pop("class")
            cls = _class_from_string(cls) if isinstance(cls, str) else cls
            plugins_instances.append(cls(**plugin))

    cache = _class_from_string(cache) if isinstance(cache, str) else cache
    instance = cache(
        serializer=serializer,
        plugins=plugins_instances,
        **kwargs)
    return instance


class CacheHandler:

    _config = {
        'default': {
            'cache': "aiocache.SimpleMemoryCache",
            'serializer': {
                'class': "aiocache.serializers.StringSerializer"
            }
        }
    }

    def __init__(self):
        self._caches = {}

    def get(self, alias):
        """
        Retrieve cache identified by alias. Will return always the same instance

        :param alias: str cache alias
        :return: cache instance
        """
        try:
            return self._caches[alias]
        except KeyError:
            pass

        config = self.get_alias_config(alias)
        cache = _create_cache(**deepcopy(config))
        self._caches[alias] = cache
        return cache

    def create(self, alias=None, cache=None, **kwargs):
        """
        Create a new cache. Either alias or cache params are required. You can use
        kwargs to pass extra parameters to configure the cache.

        :param alias: str alias to pull configuration from
        :param cache: str or class cache class to use for creating the
            new cache (when no alias is used)
        :return: New cache instance
        """
        if alias:
            config = self.get_alias_config(alias)
        elif cache:
            config = {'cache': cache}
        else:
            raise TypeError("create call needs to receive an alias or a cache")
        cache = _create_cache(**{**config, **kwargs})
        return cache

    def get_alias_config(self, alias):
        config = self.get_config()
        if alias not in config:
            raise KeyError(
                "Could not find config for '{0}', ensure you include {0} when calling"
                "caches.set_config specifying the config for that cache".format(alias))

        return config[alias]

    def get_config(self):
        """
        Return copy of current stored config
        """
        return deepcopy(self._config)

    def set_config(self, config):
        """
        Set (override) the default config for cache aliases from a dict-like structure.
        The structure is the following::

            {
                'default': {
                    'cache': "aiocache.SimpleMemoryCache",
                    'serializer': {
                        'class': "aiocache.serializers.StringSerializer"
                    }
                },
                'redis_alt': {
                    'cache': "aiocache.RedisCache",
                    'endpoint': "127.0.0.10",
                    'port': 6378,
                    'serializer': {
                        'class': "aiocache.serializers.PickleSerializer"
                    },
                    'plugins': [
                        {'class': "aiocache.plugins.HitMissRatioPlugin"},
                        {'class': "aiocache.plugins.TimingPlugin"}
                    ]
                }
            }

        'default' key must always exist when passing a new config. Default configuration
        is::

            {
                'default': {
                    'cache': "aiocache.SimpleMemoryCache",
                    'serializer': {
                        'class': "aiocache.serializers.StringSerializer"
                    }
                }
            }

        You can set your own classes there.
        The class params accept both str and class types.

        All keys in the config are optional, if they are not passed the defaults
        for the specified class will be used.

        If a config key already exists, it will be updated with the new values.
        """
        if "default" not in config:
            raise ValueError("default config must be provided")
        for config_name in config.keys():
            self._caches.pop(config_name, None)
        self._config = config


caches = CacheHandler()