# -*- coding: utf-8 -*-
"""py_deps.cache module."""
import os.path
import pickle
try:
import pylibmc
except ImportError:
pass
[docs]def backend(**kwargs):
"""Specify cache backend.
:rtype: :class:`py_deps.cache.Container`
:return: Pickle object or Memcached object.
:param kwargs: parameters
servers
Memcached servers (required in Memcached)
Required when Memcached. Using Pickle in default without ``servers``.
username
Memcached SASL username (optional)
password
Memcached SASL password (optional)
cache_name
Pickle filename (default, optional)
"""
if kwargs.get('servers'):
cache = Memcached(kwargs.get('servers'),
username=kwargs.get('username'),
password=kwargs.get('password'),
behaviors=kwargs.get('behaviors'))
else:
# default Pickle
cache = Pickle(cache_name=kwargs.get('cache_name'))
return cache
[docs]class Container:
"""Package container class."""
def __init__(self, cache_name=None):
"""Initialize."""
self.cache_name = cache_name
self.container = {}
[docs] def store_data(self, key, data):
"""Store traced_chain data."""
[docs] def read_data(self, key):
"""Read traced_chain data.
:rtype: list
:return: dependency chain list
:param tuple key: package name, version
"""
return self.container.get(key)
[docs] def list_data(self):
"""Return dictionary stored package metadata.
:rtype: dict
:return: packages metadata
"""
return self.container
[docs]class Pickle(Container):
"""Cache backend is Pickle."""
#: default cache file name
default_cache_name = 'py-deps.pickle'
def __init__(self, cache_name=None):
"""Initialize."""
if cache_name is None:
cache_name = self.default_cache_name
super().__init__(cache_name)
self.load_cache()
[docs] def load_cache(self):
"""Load cache file."""
if os.path.isfile(self.cache_name):
with open(self.cache_name, 'rb') as fobj:
self.container = pickle.load(fobj)
[docs] def save_cache(self):
"""Save cache file."""
with open(self.cache_name, 'wb') as fobj:
pickle.dump(self.container, fobj)
[docs] def store_data(self, key, data):
"""Store traced_chain data.
:param tuple key: package name, version
:param list data: traced dependency chain data
"""
self.container[key] = data
self.save_cache()
[docs]class Memcached(Container):
"""Cache backend is Memecached."""
def __init__(self, servers=None,
username=None,
password=None,
behaviors=None):
"""Initialize."""
super().__init__()
if username and password:
self.container = pylibmc.Client(servers,
binary=True,
username=username,
password=password)
else:
self.container = pylibmc.Client(servers,
binary=True)
[docs] def store_data(self, key, data):
"""Store traced_chain data.
:param tuple key: package name, version
:param list data: traced dependency chain data
"""
# pylint: disable=no-member
self.container.set('{0} {1}'.format(*key), data)
[docs] def read_data(self, key):
"""Read traced_chain data.
:rtype: list
:return: dependency chain list
:param tuple key: package name, version
"""
return self.container.get('{0} {1}'.format(*key))