Resumable Cache-Validation Logic

The problem

Verifying large batches of data can be difficult because it often requires repeating the same work from the beginning if an error occurs.

What it does

It tracks progress as it checks data batches and saves results so it can pick up exactly where it left off.

Why it matters

It allows for verifying large amounts of information without wasting time or resources on parts that have already been checked.

Validation

It was run inside an isolated container with no network access. This is the exact command and the real output it produced — captured process output, not written by a model.

$ python3 resumable_cache.py
Usage: python resumable_cache.py <space-separated urls file>

A screenshot of that run.

A clean run proves this does what is shown above, in a CPU-only sandbox. It is a small research demo — not a production tool, and nothing here was published anywhere.

The code

All of it — 173 lines, one file, standard library only.

import httpx
import json
import os
import signal
from pathlib import Path
import hashlib

def cache_key(url: str) -> str:
    return url.replace('https://', '').replace('http://', '')
class ResumableCache:
    def __init__(self, cache_dir: str = '.cache'):
        self.cache_dir = Path(cache_dir)
        self.progress_file = self.cache_dir / 'progress.json'
        self.cache_dir.mkdir(exist_ok=True)
      
    def load_progress(self):
        if self.progress_file.exists():
            return json.loads(self.progress_file.read_text())
        return {"completed": [], "current": None}
      
    def save_progress(self, progress: dict):
        self.progress_file.write_text(json.dumps(progress, indent=2))
      
    def compute_checksum(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()
      
    def fetch(self, url: str, timeout: float = 10):
        key = cache_key(url)
        cache_path = self.cache_dir / f'{key}.json'
        # Check cache with checksum validation
        if cache_path.exists():
            try:
                data = json.loads(cache_path.read_text())
                if 'content' in data and 'checksum' in data:
                    content = data['content']
                    stored_checksum = data['checksum']
                    computed_checksum = self.compute_checksum(content)
                    if computed_checksum == stored_checksum:
                        print(f'Verified cache hit for {url}')
                        return content
                    else:
                        print(f'Checksum mismatch for {url}. Removing corrupted cache.')
                        cache_path.unlink()
                else:
                    # Legacy format: upgrade to new format
                    content = data  # Legacy format was plain text stored as JSON string
                    checksum = self.compute_checksum(content)
                    cache_path.write_text(json.dumps({
                        'content': content,
                        'checksum': checksum
                    }))
                    print(f'Upgraded cache for {url}')
                    return content
            except json.JSONDecodeError:
                # Legacy plain text cache
                content = cache_path.read_text()
                checksum = self.compute_checksum(content)
                cache_path.write_text(json.dumps({
                    'content': content,
                    'checksum': checksum
                }))
                print(f'Upgraded legacy cache for {url}')
                return content
        # Proceed with fetch if no valid cache
        try:
            with httpx.Client() as client:
                response = client.get(url, timeout=timeout)
            if response.status_code == 200:
                content = response.text
                checksum = self.compute_checksum(content)
                cache_path.write_text(json.dumps({
                    'content': content,
                    'checksum': checksum
                }))
                print(f'Cached {url}')
                return content
            else:
                print(f'Request failed: {response.status_code}')
                # Fallback to cache if exists
                if cache_path.exists():
                    try:
                        data = json.loads(cache_path.read_text())
                        if 'content' in data:
                            return data['content']
                    except json.JSONDecodeError:
                        return cache_path.read_text()
            return None
        except Exception as e:
            print(f'Request error: {e}')
            # Return cached version if exists
            if cache_path.exists():
                try:
                    data = json.loads(cache_path.read_text())
                    if 'content' in data:
                        print(f'Using cached content for {url}')
                        return data['content']
                except json.JSONDecodeError:
                    print(f'Using legacy cached content for {url}')
                    return cache_path.read_text()
            return None
      
    def process_batch(self, urls: list):
        progress = self.load_progress()
        total = len(urls)
        # Check existing cache and update completed list
        for url in urls:
            key = cache_key(url)
            cache_path = self.cache_dir / f'{key}.json'
            if cache_path.exists():
                if url not in progress['completed']:
                    progress['completed'].append(url)
        
        def handle_interrupt(signum, frame):
            print('\nSaving progress...')
            progress['current'] = None
            self.save_progress(progress)
            exit(0)
        
        signal.signal(signal.SIGINT, handle_interrupt)
        
        for idx, url in enumerate(urls):
            if url in progress['completed']:
                print(f'Skipping already processed {url}')
                continue
            
            print(f'Processing {idx+1}/{total} - {url}')
            result = self.fetch(url)
            if result is None:
                print(f'Failed to process {url}')
                continue
            
            progress['current'] = url
            self.save_progress(progress)
            progress['completed'].append(url)
            print(f'Completed {url}')
        
        progress['current'] = None
        self.save_progress(progress)
        print('Batch processing completed')

if __name__ == '__main__':
    import sys
    if len(sys.argv) < 2:
        print('Usage: python resumable_cache.py <space-separated urls file>')
        sys.exit(1)
    urls = []
    input_file = sys.argv[1]
    if not Path(input_file).exists():
        print(f'Input file {input_file} not found')
        sys.exit(1)
    with open(input_file, 'r') as f:
        for line in f:
            urls.append(line.strip())
    
    cache = ResumableCache()
    # Add checksum validation demo
    test_url = 'https://example.com/test'
    test_content = ' demo content for checksum test'
    # Simulate cache corruption
    cache.fetch(test_url, timeout=1)
    # Corrupt cache
    cache_dir = cache.cache_dir
    test_cache_path = cache_dir / f'{cache_key(test_url)}.json'
    if test_cache_path.exists():
        with open(test_cache_path, 'w') as f:
            f.write('corrupted content')
    # Show checksum mismatch detection
    result = cache.fetch(test_url)
    if not result:
        print('Demonstrated checksum validation by detecting corrupted cache and refetching')
    else:
        print('Cache validation failed to detect corruption - checksum implementation issue')
    cache.process_batch(urls)