"""Module defining the handling of cache data from Overpass requests.""" import os import json import hashlib from ..constants import OSM_CACHE_DIR, OSM_TYPES def get_cache_key(query: str) -> str: """ Generate a unique cache key for the query using a hash function. This ensures that queries with different parameters are cached separately. """ return hashlib.md5(query.encode('utf-8')).hexdigest() class CachingStrategyBase: """ Base class for implementing caching strategies. """ def get(self, key): """Retrieve the cached data associated with the provided key.""" raise NotImplementedError('Subclass should implement get') def set(self, key, value): """Store data in the cache with the specified key.""" raise NotImplementedError('Subclass should implement set') def set_hollow(self, key, **kwargs): """Create a hollow (empty) cache entry with a specific key.""" raise NotImplementedError('Subclass should implement set_hollow') def close(self): """Clean up or close any resources used by the caching strategy.""" class JSONCache(CachingStrategyBase): """ A caching strategy that stores and retrieves data in JSON format. """ def __init__(self, cache_dir=OSM_CACHE_DIR): # Add the class name as a suffix to the directory self._cache_dir = f'{cache_dir}' if not os.path.exists(self._cache_dir): os.makedirs(self._cache_dir) def _filename(self, key): return os.path.join(self._cache_dir, f'{key}.json') def get(self, key): """Retrieve JSON data from the cache and parse it as an ElementTree.""" filename = self._filename(key) if os.path.exists(filename): try: # Open and parse the cached JSON data with open(filename, 'r', encoding='utf-8') as file: data = json.load(file) # Return the data as a list of dicts. return data except json.JSONDecodeError: return None # Return None if parsing fails return None def set(self, key, value): """Save the JSON data in the cache.""" filename = self._filename(key) try: # Write the JSON data to the cache file with open(filename, 'w', encoding='utf-8') as file: json.dump(value, file, ensure_ascii=False, indent=4) except IOError as e: raise IOError(f"Error writing to cache file: {filename} - {e}") from e def set_hollow(self, key, cell: tuple, osm_types: list, selector: str, conditions: list=None, out='center'): """Create an empty placeholder cache entry for a future fill.""" hollow_key = f'hollow_{key}' filename = self._filename(hollow_key) # Create the hollow JSON structure hollow_data = { "key": key, "cell": list(cell), "osm_types": list(osm_types), "selector": selector, "conditions": conditions, "out": out } # Write the hollow data to the cache file try: with open(filename, 'w', encoding='utf-8') as file: json.dump(hollow_data, file, ensure_ascii=False, indent=4) except IOError as e: raise IOError(f"Error writing hollow cache to file: {filename} - {e}") from e def close(self): """Cleanup method, if needed.""" class CachingStrategy: """ A class to manage different caching strategies. """ __strategy = JSONCache() # Default caching strategy __strategies = { 'JSON': JSONCache, } @classmethod def use(cls, strategy_name='JSON', **kwargs): """Define the caching strategy to use.""" if cls.__strategy: cls.__strategy.close() strategy_class = cls.__strategies.get(strategy_name) if not strategy_class: raise ValueError(f"Unknown caching strategy: {strategy_name}") cls.__strategy = strategy_class(**kwargs) return cls.__strategy @classmethod def get(cls, key): """Get the data from the cache.""" return cls.__strategy.get(key) @classmethod def set(cls, key, value): """Save the data in the cache.""" cls.__strategy.set(key, value) @classmethod def set_hollow(cls, key, cell: tuple, osm_types: OSM_TYPES, selector: str, conditions: list=None, out='center'): """Create a hollow cache entry.""" cls.__strategy.set_hollow(key, cell, osm_types, selector, conditions, out)