Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ########################################################################################################################
- # File: cache.py
- # Author: Dan Huckson, https://github.com/unodan
- # Date: 2019-09-07
- ########################################################################################################################
- from os import path, makedirs
- from json import load, dump
- from copy import deepcopy
- from importlib.util import module_from_spec, spec_from_file_location
- def uri2dict(uri, value):
- node = output = {}
- parts = uri.split('/')
- while parts:
- _path = parts.pop(0)
- node[_path] = node = {}
- c = Cache(**output)
- c.set(uri, value)
- return c.get()
- class Cache:
- def __init__(self, *args, **kwargs):
- self.nodes = {}
- arg = args[0] if args else kwargs
- if arg:
- key = next(iter(arg.keys()))
- value = next(iter(arg.values()))
- if '/' in key:
- self.set(key, value)
- else:
- self.nodes = arg
- self.indent = '.'
- def __eq__(self, other):
- if self.nodes == other.nodes:
- return True
- return False
- def __str__(self):
- return str(self.nodes.items())
- def __repr__(self):
- data = "**{'key': 'value'}"
- return f"{self.__class__.__name__}({data})"
- def get(self, uri=None):
- if not uri:
- return self.nodes
- def walk(_uri, nodes):
- parts = _uri.split('/', 1)
- key = parts.pop(0)
- if key in nodes:
- node = nodes[key]
- if not parts:
- return node
- else:
- return walk(parts[0], node)
- return walk(uri, self.nodes)
- def set(self, uri, *args, **kwargs):
- data = args[0] if args else kwargs
- def walk(_uri, nodes):
- parts = _uri.split('/', 1)
- key = parts.pop(0)
- if key in nodes and parts:
- walk(parts[0], nodes[key])
- elif len(_uri.split('/')) == 1:
- nodes[key] = data
- else:
- self.nodes.update(**uri2dict(_uri, data))
- walk(uri, self.nodes)
- def dump(self, indent=None):
- """ Dumps the contents of the cache to the screen.
- The output from dump goes stdout and is used to view the cache contents.
- Default indentation is a dot for each level.
- :param indent:
- indent (str): String to be use for indenting levels.
- :return:
- Nothing.
- """
- indent = indent if indent else '.'
- print('-------------------------------------------------------------------------------------------------------')
- print('id =', id(self), '\nnodes =', self)
- if self.nodes:
- def walk(_cfg, count):
- count += 1
- for key, value in _cfg.items():
- if isinstance(value, dict):
- print(indent * count, key)
- walk(value, count)
- else:
- if isinstance(value, str):
- value = f'"{value}"'
- print(indent * count, key, f'value={value}')
- walk(self.nodes, 0)
- else:
- print(' (No Data)')
- print('-------------------------------------------------------------------------------------------------------')
- def save(self, file=None):
- if file:
- dirname = path.dirname(file)
- if dirname and not path.exists(dirname):
- makedirs(dirname)
- with open(file, 'w') as f:
- dump(self.nodes, f, indent=3)
- def load(self, file=None):
- file_type = path.splitext(file)[1].lstrip('.').lower()
- if file_type == 'py' and path.exists(file):
- spec = spec_from_file_location("module.name", file)
- module = module_from_spec(spec)
- spec.loader.exec_module(module)
- self.nodes = module.config
- if file_type == 'json' and path.exists(file):
- with open(file) as f:
- self.nodes = load(f)
- def copy(self):
- return Cache(**deepcopy(self.nodes))
- def remove(self, uri):
- """ Remove entree from cache.
- Removes an entree from the cache if it exists.
- :param uri:
- uri (str): URI that points to the entree to remove.
- :return:
- Nothing.
- """
- uri = uri.lstrip('/')
- if self.exists(uri):
- node = self.get('/'.join(uri.split('/')[:-1]))
- del node[uri.split('/')[-1]]
- def exists(self, uri):
- """ Test if URI exists in the cache.
- :param uri:
- :return:
- """
- return True if self.get(uri) else False
- def destroy(self):
- """ Destroy cache.
- Deletes all entries in the cache.
- :return:
- Nothing.
- """
- del self.nodes
- self.nodes = {}
- def has_nodes(self):
- if len(self.nodes):
- return True
- return False
- # If uncommented the cache will ba a singleton.
- # class Cache:
- # instance = None
- #
- # def __new__(cls, **kwargs):
- # if not cls.instance:
- # cls.instance = super(Cache, cls).__new__(cls)
- # cls.instance.cache = _Cache(**kwargs)
- #
- # return cls.instance.cache
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement