| from datetime import datetime |
| from datetime import timedelta |
| import json |
| |
| from recipe_engine import recipe_api |
| |
| JSON_ENTRY_NAME = 'name' |
| JSON_ENTRY_UPDATED_DATE = 'updated_date' |
| JSON_ENTRY_REMOVAL_DATE = 'removal_date' |
| |
| PACKAGE_REMOVAL_INTERVAL_DAYS = 30 |
| |
| |
| class CacheEntry: |
| """An object used to represent a file and its usage in a cache.""" |
| |
| def __init__( |
| self, |
| file_name: str, |
| updated_date: datetime = None, |
| removal_date: datetime = None, |
| ): |
| self.name = file_name |
| |
| # The updated date will be stored as a string for easy conversion to and from json. |
| self.updated_date = updated_date |
| self.updated_date = updated_date.strftime("%m/%d/%Y, %H:%M:%S") |
| |
| # The removal date will be stored as a string for easy conversion to and from json. |
| self.removal_date = removal_date |
| if self.removal_date is None: |
| self._calc_removal_date(self.updated_date) |
| else: |
| self.removal_date = removal_date.strftime("%m/%d/%Y, %H:%M:%S") |
| |
| def _calc_removal_date(self, updated_date: datetime): |
| """Calculate the removal date as the update date + 30 days.""" |
| time_delta = timedelta(days=PACKAGE_REMOVAL_INTERVAL_DAYS) |
| updated_date_datetime = datetime.strptime( |
| updated_date, "%m/%d/%Y, %H:%M:%S" |
| ) |
| new_removal_date = updated_date_datetime + time_delta |
| self.removal_date = new_removal_date.strftime("%m/%d/%Y, %H:%M:%S") |
| |
| def removal_date_as_datetime(self) -> datetime: |
| """Return the stored removal timestamp string as a datetime object.""" |
| return self._convert_str_to_datetime(self.removal_date) |
| |
| def _convert_str_to_datetime(self, date_str: str) -> datetime: |
| """Return the converted string date as a datetime object""" |
| return datetime.strptime(date_str, "%m/%d/%Y, %H:%M:%S") |
| |
| |
| class CacheMicroManagerApi(recipe_api.RecipeApi): |
| """This module keeps track of root individual files within a directory (based on |
| a non-recursivce mode). It keeps track of the last time a file was used and |
| proactively cleans the directory by removing files older than some specified time. |
| """ |
| |
| # pylint: disable=unused-argument |
| def __init__(self, *args, **kwargs): |
| """Create a new CacheMicroManager object. |
| |
| Args: |
| target_dir (str): the path to the cache directory containing dependencies. |
| """ |
| super(CacheMicroManagerApi, self).__init__(*args, **kwargs) |
| self.cache_target_directory = None |
| self.cache_name = None |
| self.metadata_file_name = None |
| self.cache_metadata_file_path = None |
| |
| def _initialize(self, target_dir): |
| self.cache_target_directory = self.m.path.cast_to_path(target_dir) |
| self.cache_name = self.m.path.basename(self.cache_target_directory) |
| self.metadata_file_name = '.{}_cache_metadata.json'.format(self.cache_name) |
| self.cache_metadata_file_path = ( |
| self.cache_target_directory / self.metadata_file_name |
| ) |
| |
| def today(self): |
| """Provide a deterministic date when running tests.""" |
| return datetime.now() # pragma: nocover |
| |
| def run(self, target_dir, deps_list: list): |
| """Run the cache micro manager on the target directory. |
| |
| If the directory is not yet being tracked by the cache micro manager it will read the contents of |
| the target_dir_path and all current files and directories to cache metadata file. |
| |
| If the directory is being tracked it will add any dependencies in deps_list not present in the file |
| to the metadata file or update the updated and removal date of dependencies in deps_list in the file. |
| |
| After the deps are updated or added it will perform the following minor cleanup: |
| * If there is an entry in the metadata file but the package is not on disk it will remove the entry |
| from the file. |
| * If there is a package on disk but not in the file it will simply add it to the metadata file for |
| tracking and eventual removal. |
| * Then it will look at all removal dates from the metadata file and delete all expired packages from |
| disk and remove them from the metadata file. |
| |
| Args: |
| * deps_list(list[str]): the list of dependencies that are currently being used. |
| """ |
| |
| self._initialize(target_dir=target_dir) |
| |
| with self.m.step.nest('Running Cache Micro Manager on {}.'.format( |
| self.cache_target_directory)): |
| |
| cache_exists = self.m.path.exists(self.cache_target_directory) |
| if not cache_exists: |
| self.m.step.empty( |
| 'Cache Micro Manager, cache directory exists check', |
| status='SUCCESS', |
| step_text='Cache dir does not exist, skipping.' |
| ) |
| return |
| |
| if not self.m.path.exists(self.cache_metadata_file_path): |
| directory_files_list = self.m.file.listdir( |
| 'Reading cache directory {}'.format(self.cache_target_directory), |
| self.cache_target_directory, |
| recursive=False, |
| ) |
| |
| directory_cache_entry_list = self.convert_file_list_to_cache_entry_list( |
| directory_files_list |
| ) |
| |
| if len(directory_cache_entry_list) > 0: |
| # It should be safe to ignore the check for existence of the file since no one |
| # actively logs onto the bots and manipulates the file system. |
| self.m.file.write_text( |
| 'Writing cache metadata file.', |
| self.cache_metadata_file_path, |
| json.dumps([ob.__dict__ for ob in directory_cache_entry_list]), |
| ) |
| else: |
| # the currently stored metadata entries. |
| |
| # example of the data that is stored in the metadata file. |
| # [ |
| # { |
| # "name": "package_1", |
| # "updated_date": "datetime", |
| # "removal_date": "datetime + interval" |
| # }, |
| # { |
| # "name": "package_2", |
| # "updated_date": "datetime", |
| # "removal_date": "datetime + interval" |
| # } |
| # ] |
| |
| current_metadata_entries = self.read_metadata_file() |
| |
| # these files may not be in the metadata file, no telling how they got there. |
| current_directory_entries = self.m.file.listdir( |
| 'Reading cache directory {}'.format(self.cache_target_directory), |
| self.cache_target_directory, False |
| ) |
| |
| # remove the cache file itself so we do not record it in the file. |
| current_directory_entries.remove(self.cache_metadata_file_path) |
| |
| # add or update the current working dependencies. |
| for dep in deps_list: |
| if not self.is_file_name_in_cache_entry_list( |
| dep, current_metadata_entries): |
| # we need to add it. |
| current_metadata_entries.append( |
| CacheEntry(file_name=str(dep), updated_date=self.today()) |
| ) |
| else: |
| # we want to update it. |
| existing_entry = self.get_cache_entry_from_list( |
| str(dep), current_metadata_entries |
| ) |
| |
| # package will be active for another month. |
| new_entry = CacheEntry( |
| file_name=str(existing_entry.name), updated_date=self.today() |
| ) |
| current_metadata_entries.remove(existing_entry) |
| current_metadata_entries.append(new_entry) |
| |
| paths_as_str_current_dir_entries = [ |
| str(item) for item in current_directory_entries |
| ] |
| # there can be deps in the file and not in the directory or |
| for current_meta_entry in current_metadata_entries: # current metadata entries should be str |
| if str(current_meta_entry.name |
| ) not in paths_as_str_current_dir_entries: |
| current_metadata_entries.remove(current_meta_entry) |
| |
| # there can be deps in the directory and not in the file. |
| for file_path in paths_as_str_current_dir_entries: |
| cacheEntry = self.get_cache_entry_from_list( |
| str(file_path), current_metadata_entries |
| ) |
| if cacheEntry is None: |
| current_metadata_entries.append( |
| CacheEntry(file_name=str(file_path), updated_date=self.today()) |
| ) |
| |
| # Check dates and delete unused packages. |
| for cacheEntry in current_metadata_entries: |
| today = self.today() |
| if cacheEntry.removal_date_as_datetime().date() < today.date(): |
| self.delete_file(cacheEntry.name) |
| current_metadata_entries.remove(cacheEntry) |
| |
| # write the new file contents. |
| self.m.file.remove( |
| 'Removing existing cache file {}'.format( |
| self.cache_metadata_file_path |
| ), self.cache_metadata_file_path |
| ) |
| self.m.file.write_text( |
| 'Writing cache metadata file.', |
| self.cache_metadata_file_path, |
| json.dumps([ob.__dict__ for ob in current_metadata_entries]), |
| ) |
| |
| def get_cache_entry_from_list( |
| self, file_name: str, cache_entry_list: list |
| ) -> CacheEntry: |
| """Get the associated cache entry when supplied with a specific file name. |
| |
| Args: |
| * file_name (str): the name of a file/directory that represents a dependency. |
| * cache_entry_list (list[CacheEntry]): a list of CacheEntry objects. |
| |
| Returns: |
| CacheEntry | None: Returns a CacheEntry if found in the list otherwise returns None. |
| """ |
| for cache_entry in cache_entry_list: |
| if file_name == cache_entry.name: |
| return cache_entry |
| return None |
| |
| # file name is a path as passed in here |
| def is_file_name_in_cache_entry_list( |
| self, file_name, cache_entry_list: list |
| ) -> bool: |
| """Check to see if a CacheEntry is in the list for the supplied file_name. |
| |
| Args: |
| * file_name (str): the name of a file/directory that represents a dependency. |
| * cache_entry_list (list[CacheEntry]): a list of CacheEntry objects. |
| |
| Returns: |
| bool: True if a cache entry exists with file_name as its name, False if it does not. |
| """ |
| for cache_entry in cache_entry_list: |
| if str(file_name) == str(cache_entry.name): |
| return True |
| return False |
| |
| def convert_file_list_to_cache_entry_list( |
| self, file_names_list: list |
| ) -> list: |
| """Create a list of CacheEntry objects from a list of file names. |
| |
| The primary usage of this is on start of CacheMicroManager when it has not yet begun |
| to manage a directory and we need to record the contents of it. |
| |
| Args: |
| * file_names_list (list[str]): a list of file names to use to instantiate CacheEntry |
| objects. |
| |
| Returns: |
| list[CacheEntry]: a list of newly instantiated CacheEntry objects associate with the |
| supplied file names list. |
| """ |
| cache_entry_list = [] |
| for fd in file_names_list: |
| cache_entry = CacheEntry(file_name=str(fd), updated_date=self.today()) |
| cache_entry_list.append(cache_entry) |
| return cache_entry_list |
| |
| def delete_file(self, file_name): |
| """Delete a file or directory. |
| |
| Args: |
| * file_name (Path): a file or directory name that will be deleted. |
| """ |
| try: |
| if self.m.path.isfile(file_name): |
| self.m.file.remove( |
| 'Removing file descriptor {}'.format(file_name), file_name |
| ) #pragma: nocover |
| elif self.m.path.isdir(file_name): |
| try: #pragma: nocover |
| self.m.file.rmtree( |
| 'Removig dir file descriptor {} (rmtree)'.format(file_name), |
| file_name |
| ) #pragma: nocover |
| except self.m.file.Error: #pragma: nocover |
| self.m.step( |
| 'Removing dir file descriptor {} (fallback rm)'.format(file_name), |
| ['rm', '-rf', file_name] |
| ) #pragma: nocover |
| except self.m.file.Error: #pragma: nocover |
| print('File not found.') #pragma: nocover |
| |
| def read_metadata_file(self) -> list: |
| """Read the metadata file at the self.cache_metadata_file_location path. |
| |
| Returns: |
| list[CacheEntry]: returns the list of CacheEntry's found in the existing file. |
| """ |
| with self.m.step.nest('Reading metadata file {}'.format( |
| self.cache_metadata_file_path)): |
| # it is possible to return an empty list, populated list or None. |
| meta_json = self.m.file.read_text( |
| 'Reading {}'.format(self.cache_metadata_file_path), |
| self.cache_metadata_file_path, |
| ) |
| meta_cache_entries = [] |
| json_items = json.loads(meta_json) |
| for item in json_items: |
| # TODO(ricardoamador) remove this code once the old dates without times are removed. |
| updated_date_found = None |
| if (self.date_format_check(item[JSON_ENTRY_UPDATED_DATE])): |
| updated_date_found = datetime.strptime( |
| item[JSON_ENTRY_UPDATED_DATE], "%m/%d/%Y, %H:%M:%S" |
| ) |
| else: |
| updated_date_found = datetime.strptime( |
| "{}, 12:00:00".format(item[JSON_ENTRY_UPDATED_DATE]), |
| "%m/%d/%Y, %H:%M:%S" |
| ) |
| |
| removal_date_found = None |
| if (self.date_format_check(item[JSON_ENTRY_REMOVAL_DATE])): |
| removal_date_found = datetime.strptime( |
| item[JSON_ENTRY_REMOVAL_DATE], "%m/%d/%Y, %H:%M:%S" |
| ) |
| else: |
| removal_date_found = datetime.strptime( |
| "{}, 12:00:00".format(item[JSON_ENTRY_REMOVAL_DATE]), |
| "%m/%d/%Y, %H:%M:%S" |
| ) |
| |
| new_entry = CacheEntry( |
| file_name=str(item[JSON_ENTRY_NAME]), |
| updated_date=updated_date_found, |
| removal_date=removal_date_found |
| ) |
| meta_cache_entries.append(new_entry) |
| return meta_cache_entries |
| |
| def date_format_check(self, date_str: str) -> bool: |
| """Check the date format of the date we are attempting to process. |
| |
| returns: |
| True if the time stamp includes the time and the date. False if non-compliant. |
| """ |
| try: |
| datetime.strptime(date_str, "%m/%d/%Y, %H:%M:%S") |
| return True |
| except ValueError: |
| return False |
