json_disk_cache.py

def disk_cache(cache_dir: str = ".cache"):
    """
    Decorator that implements disk caching for functions using JSON.
    NOTE: This is adapted for async usage: we await the wrapped function,
    but the file I/O remains synchronous for simplicity.
    Args:
        cache_dir: Directory to store cache files
    """

    def decorator(func):
        # Create cache directory if it doesn't exist
        cache_path = Path(cache_dir)
        cache_path.mkdir(parents=True, exist_ok=True)

        @wraps(func)
        async def wrapper(*args, **kwargs):
            # Create a unique key from the function arguments
            key_dict = {
                "args": str(args),  # Convert args to string for JSON compatibility
                "kwargs": str(
                    kwargs
                ),  # Convert kwargs to string for JSON compatibility
                "func_name": func.__name__,
            }
            key_bytes = json.dumps(key_dict, sort_keys=True).encode("utf-8")
            key_hash = hashlib.sha256(key_bytes).hexdigest()

            cache_file = cache_path / f"{key_hash}.json"

            # Try to load from cache
            if cache_file.exists():
                try:
                    with open(cache_file, "r", encoding="utf-8") as f:
                        return json.load(f)
                except json.JSONDecodeError:
                    # If cache is corrupted, remove it
                    cache_file.unlink(missing_ok=True)

            # If not in cache or cache is corrupted, compute the result
            result = await func(*args, **kwargs)

            # Save to cache
            try:
                with open(cache_file, "w", encoding="utf-8") as f:
                    json.dump(result, f, ensure_ascii=False, indent=2)
            except (TypeError, OSError) as e:
                print(f"Warning: Failed to cache result: {e}")

            return result

        def clear_cache():
            """Clear all cached results for this function"""
            for cf in cache_path.glob("*.json"):
                cf.unlink(missing_ok=True)

        wrapper.clear_cache = clear_cache
        return wrapper

    return decorator