Source code for coalib.misc.CachingUtilities

import hashlib
import logging
import os
import pickle

from coalib.misc import Constants


[docs]def get_data_path(log_printer, identifier): """ Get the full path of ``identifier`` present in the user's data directory. :param log_printer: A LogPrinter object to use for logging. :param identifier: The file whose path needs to be expanded. :return: Full path of the file, assuming it's present in the user's config directory. Returns ``None`` if there is a ``PermissionError`` in creating the directory. """ try: os.makedirs(Constants.USER_DATA_DIR, exist_ok=True) return os.path.join(Constants.USER_DATA_DIR, hash_id(identifier)) except PermissionError: logging.error("Unable to create user data directory '{}'. Continuing" ' without caching.'.format(Constants.USER_DATA_DIR)) return None
[docs]def delete_files(log_printer, identifiers): """ Delete the given identifiers from the user's coala data directory. :param log_printer: A LogPrinter object to use for logging. :param identifiers: The list of files to be deleted. :return: True if all the given files were successfully deleted. False otherwise. """ error_files = [] result = True for identifier in identifiers: try: file_path = get_data_path(None, identifier) if os.path.isfile(file_path): os.remove(file_path) else: result = False except (OSError, TypeError) as e: error_files.append(hash_id(identifier)) if len(error_files) > 0: error_files = ', '.join(error_files) logging.warning('There was a problem deleting the following ' 'files: {}. Please delete them manually from ' "'{}'.".format(error_files, Constants.USER_DATA_DIR)) result = False return result
[docs]def pickle_load(log_printer, identifier, fallback=None): """ Get the data stored in ``filename`` present in the user config directory. Example usage: >>> test_data = {'answer': 42} >>> pickle_dump(None, 'test_project', test_data) True >>> pickle_load(None, 'test_project') {'answer': 42} >>> pickle_load(None, 'nonexistent_project') >>> pickle_load(None, 'nonexistent_project', fallback=42) 42 :param log_printer: A LogPrinter object to use for logging. :param identifier: The name of the file present in the user config directory. :param fallback: Return value to fallback to in case the file doesn't exist. :return: Data that is present in the file, if the file exists. Otherwise the ``default`` value is returned. """ file_path = get_data_path(None, identifier) if file_path is None or not os.path.isfile(file_path): return fallback with open(file_path, 'rb') as f: try: return pickle.load(f) except (pickle.UnpicklingError, EOFError) as e: logging.warning('The given file is corrupted and will be removed.') delete_files(None, [identifier]) return fallback
[docs]def pickle_dump(log_printer, identifier, data): """ Write ``data`` into the file ``filename`` present in the user config directory. :param log_printer: A LogPrinter object to use for logging. :param identifier: The name of the file present in the user config directory. :param data: Data to be serialized and written to the file using pickle. :return: True if the write was successful. False if there was a permission error in writing. """ file_path = get_data_path(None, identifier) if file_path is None: # Exit silently since the error has been logged in ``get_data_path`` return False with open(file_path, 'wb') as f: pickle.dump(data, f) return True
[docs]def hash_id(text): """ Hashes the given text. :param text: String to to be hashed :return: A MD5 hash of the given string """ return hashlib.md5(text.encode('utf-8')).hexdigest()
[docs]def get_settings_hash(sections, targets=[], ignore_settings: list = ['disable_caching']): """ Compute and return a unique hash for the settings. :param sections: A dict containing the settings for each section. :param targets: The list of sections that are enabled. :param ignore_settings: Setting keys to remove from sections before hashing. :return: A MD5 hash that is unique to the settings used. """ settings = [] for section in sections: if section in targets or targets == []: section_copy = sections[section].copy() for setting in ignore_settings: try: section_copy.__getitem__(setting, ignore_defaults=True) section_copy.delete_setting(setting) except IndexError: continue settings.append(str(section_copy)) return hash_id(str(settings))
[docs]def settings_changed(log_printer, settings_hash): """ Determine if the settings have changed since the last run with caching. :param log_printer: A LogPrinter object to use for logging. :param settings_hash: A MD5 hash that is unique to the settings used. :return: Return True if the settings hash has changed Return False otherwise. """ project_hash = hash_id(os.getcwd()) settings_hash_db = pickle_load(None, 'settings_hash_db', {}) if project_hash not in settings_hash_db: # This is the first time coala is run on this project, so the cache # will be flushed automatically. return False result = settings_hash_db[project_hash] != settings_hash if result: del settings_hash_db[project_hash] logging.debug('Since the configuration settings have changed since ' 'the last run, the cache will be flushed and rebuilt.') return result
[docs]def update_settings_db(log_printer, settings_hash): """ Update the config file last modification date. :param log_printer: A LogPrinter object to use for logging. :param settings_hash: A MD5 hash that is unique to the settings used. """ project_hash = hash_id(os.getcwd()) settings_hash_db = pickle_load(None, 'settings_hash_db', {}) settings_hash_db[project_hash] = settings_hash pickle_dump(None, 'settings_hash_db', settings_hash_db)