Module coscine.cache
This file implements a basic persistent cache for nonvolatile Coscine data such as metadata vocabularies. The data is automatically refreshed every once in a while.
Expand source code
###############################################################################
# Coscine Python SDK
# Copyright (c) 2018-2022 RWTH Aachen University
# Licensed under the terms of the MIT License
# #############################################################################
# Coscine, short for Collaborative Scientific Integration Environment is
# a platform for research data management (RDM).
# For more information on Coscine visit https://www.coscine.de/.
#
# Please note that this python module is open source software primarily
# developed and maintained by the scientific community. It is not
# an official service that RWTH Aachen provides support for.
###############################################################################
###############################################################################
# File description
###############################################################################
"""
This file implements a basic persistent cache for nonvolatile Coscine
data such as metadata vocabularies.
The data is automatically refreshed every once in a while.
"""
###############################################################################
# Dependencies
###############################################################################
from __future__ import annotations
from .__about__ import __version__
from .defaults import APPDIR, CACHEFILE, TIMEFORMAT
import atexit
import json
import os
from json.decoder import JSONDecodeError
from datetime import datetime, timedelta
###############################################################################
# Class
###############################################################################
class Cache:
"""
A basic persistent cache for temporary storage of nonvolatile data.
Data is automatically refreshed when it exceeds a certain age.
Data is loaded from & stored on disk if persistency option is enabled.
Attributes
-----------
_cache : dict
A simple dictionary to store the cached data.
_persistent : bool
Controls cache persistence - if enabled the cache is saved to
a file and restored upon the next session.
"""
_cache: dict
_persistent: bool
###############################################################################
def __init__(self, persistent: bool = True) -> None:
"""
Initializes a Cache instance and attempts to load a previous
cache copy from a file if the $persistent option is set.
Parameters
----------
persistent : bool
Enable to save the cache in a file upon program exit and
to restore it in the next session.
"""
self._persistent = persistent
self.load()
###############################################################################
def save(self) -> None:
"""
Saves the current session cache into a file if
the $persistent option is set (see __init__()).
Any cachefile existing prior is overwritten.
"""
if self._persistent and self._cache:
path = APPDIR.user_cache_dir
if not os.path.exists(path):
os.makedirs(path, exist_ok=True)
filepath = os.path.join(path, CACHEFILE)
fd = open(filepath, "w")
fd.write(json.dumps(self._cache))
fd.close()
###############################################################################
def load(self) -> None:
"""
Loads a previously stored session cache for use with the
current session. If no previous session is found, the cache
is intialized as an empty dict.
"""
if self._persistent:
atexit.register(self.save)
try:
path = APPDIR.user_cache_dir
filepath = os.path.join(path, CACHEFILE)
fd = open(filepath, "r")
self._cache = json.loads(fd.read())
fd.close()
except (FileNotFoundError, JSONDecodeError):
self._cache = {}
else:
self._cache = {}
###############################################################################
def get(self, key: str) -> str:
"""
Attempts to read a cached dataset from the cache via a key.
Parameters
----------
key : str
The key used to save data in the cache (e.g. a URL)
Returns
--------
str
The cached data for the given key
None
If the key is not present in the cache or if the data is
likely to be outdated
"""
if key in self._cache:
lastTime = datetime.strptime(self._cache[key]["time"], TIMEFORMAT)
if ((datetime.now() - lastTime) < timedelta(days=24)):
return self._cache[key]["data"]
return None
###############################################################################
def set(self, key: str, data) -> None:
"""
Sets data inside the cache via a key. If that key is already
present any data referenced by it is overwritten.
Parameters
----------
key : str
The key used for saving data in the cache (e.g. a URL)
"""
self._cache[key] = {
"time": datetime.now().strftime(TIMEFORMAT),
"data": data
}
###############################################################################
def clear(self) -> None:
"""
Clears the cache
"""
self._cache.clear()
###############################################################################Classes
class Cache (persistent: bool = True)-
A basic persistent cache for temporary storage of nonvolatile data. Data is automatically refreshed when it exceeds a certain age. Data is loaded from & stored on disk if persistency option is enabled.
Attributes
_cache:dict- A simple dictionary to store the cached data.
_persistent:bool- Controls cache persistence - if enabled the cache is saved to a file and restored upon the next session.
Initializes a Cache instance and attempts to load a previous cache copy from a file if the $persistent option is set.
Parameters
persistent:bool- Enable to save the cache in a file upon program exit and to restore it in the next session.
Expand source code
class Cache: """ A basic persistent cache for temporary storage of nonvolatile data. Data is automatically refreshed when it exceeds a certain age. Data is loaded from & stored on disk if persistency option is enabled. Attributes ----------- _cache : dict A simple dictionary to store the cached data. _persistent : bool Controls cache persistence - if enabled the cache is saved to a file and restored upon the next session. """ _cache: dict _persistent: bool ############################################################################### def __init__(self, persistent: bool = True) -> None: """ Initializes a Cache instance and attempts to load a previous cache copy from a file if the $persistent option is set. Parameters ---------- persistent : bool Enable to save the cache in a file upon program exit and to restore it in the next session. """ self._persistent = persistent self.load() ############################################################################### def save(self) -> None: """ Saves the current session cache into a file if the $persistent option is set (see __init__()). Any cachefile existing prior is overwritten. """ if self._persistent and self._cache: path = APPDIR.user_cache_dir if not os.path.exists(path): os.makedirs(path, exist_ok=True) filepath = os.path.join(path, CACHEFILE) fd = open(filepath, "w") fd.write(json.dumps(self._cache)) fd.close() ############################################################################### def load(self) -> None: """ Loads a previously stored session cache for use with the current session. If no previous session is found, the cache is intialized as an empty dict. """ if self._persistent: atexit.register(self.save) try: path = APPDIR.user_cache_dir filepath = os.path.join(path, CACHEFILE) fd = open(filepath, "r") self._cache = json.loads(fd.read()) fd.close() except (FileNotFoundError, JSONDecodeError): self._cache = {} else: self._cache = {} ############################################################################### def get(self, key: str) -> str: """ Attempts to read a cached dataset from the cache via a key. Parameters ---------- key : str The key used to save data in the cache (e.g. a URL) Returns -------- str The cached data for the given key None If the key is not present in the cache or if the data is likely to be outdated """ if key in self._cache: lastTime = datetime.strptime(self._cache[key]["time"], TIMEFORMAT) if ((datetime.now() - lastTime) < timedelta(days=24)): return self._cache[key]["data"] return None ############################################################################### def set(self, key: str, data) -> None: """ Sets data inside the cache via a key. If that key is already present any data referenced by it is overwritten. Parameters ---------- key : str The key used for saving data in the cache (e.g. a URL) """ self._cache[key] = { "time": datetime.now().strftime(TIMEFORMAT), "data": data } ############################################################################### def clear(self) -> None: """ Clears the cache """ self._cache.clear()Methods
def clear(self) ‑> None-
Clears the cache
Expand source code
def clear(self) -> None: """ Clears the cache """ self._cache.clear() def get(self, key: str) ‑> str-
Attempts to read a cached dataset from the cache via a key.
Parameters
key:str- The key used to save data in the cache (e.g. a URL)
Returns
str- The cached data for the given key
None- If the key is not present in the cache or if the data is likely to be outdated
Expand source code
def get(self, key: str) -> str: """ Attempts to read a cached dataset from the cache via a key. Parameters ---------- key : str The key used to save data in the cache (e.g. a URL) Returns -------- str The cached data for the given key None If the key is not present in the cache or if the data is likely to be outdated """ if key in self._cache: lastTime = datetime.strptime(self._cache[key]["time"], TIMEFORMAT) if ((datetime.now() - lastTime) < timedelta(days=24)): return self._cache[key]["data"] return None def load(self) ‑> None-
Loads a previously stored session cache for use with the current session. If no previous session is found, the cache is intialized as an empty dict.
Expand source code
def load(self) -> None: """ Loads a previously stored session cache for use with the current session. If no previous session is found, the cache is intialized as an empty dict. """ if self._persistent: atexit.register(self.save) try: path = APPDIR.user_cache_dir filepath = os.path.join(path, CACHEFILE) fd = open(filepath, "r") self._cache = json.loads(fd.read()) fd.close() except (FileNotFoundError, JSONDecodeError): self._cache = {} else: self._cache = {} def save(self) ‑> None-
Saves the current session cache into a file if the $persistent option is set (see init()). Any cachefile existing prior is overwritten.
Expand source code
def save(self) -> None: """ Saves the current session cache into a file if the $persistent option is set (see __init__()). Any cachefile existing prior is overwritten. """ if self._persistent and self._cache: path = APPDIR.user_cache_dir if not os.path.exists(path): os.makedirs(path, exist_ok=True) filepath = os.path.join(path, CACHEFILE) fd = open(filepath, "w") fd.write(json.dumps(self._cache)) fd.close() def set(self, key: str, data) ‑> None-
Sets data inside the cache via a key. If that key is already present any data referenced by it is overwritten.
Parameters
key:str- The key used for saving data in the cache (e.g. a URL)
Expand source code
def set(self, key: str, data) -> None: """ Sets data inside the cache via a key. If that key is already present any data referenced by it is overwritten. Parameters ---------- key : str The key used for saving data in the cache (e.g. a URL) """ self._cache[key] = { "time": datetime.now().strftime(TIMEFORMAT), "data": data }