Python Cache¶

import time
import threading
from typing import Any, Optional

class PythonCache:
    def __init__(self, default_expiration: Optional[int] = None):
        """
        A pure Python in-memory cache implementation.

        Args:
            default_expiration (Optional[int]): Default expiration time in seconds for keys.
                                                If None, keys do not expire unless specified.
        """
        self.cache = {}  # To store cache items: {key: (value, expiration_timestamp)}
        self.default_expiration = default_expiration
        self.lock = threading.Lock()  # To manage thread-safe operations

        # Start a background thread for garbage collection
        self.gc_thread = threading.Thread(target=self._garbage_collector, daemon=True)
        self.gc_thread.start()

    def _current_time(self) -> float:
        """Gets the current time."""
        return time.time()

    def set(self, key: str, value: Any, expiration: Optional[int] = None):
        """
        Set a key-value pair in the cache.

        Args:
            key (str): The key to store.
            value (Any): The value to store.
            expiration (Optional[int]): Expiration time in seconds. If None, uses the default expiration.
        """
        with self.lock:
            expire_at = self._current_time() + expiration if expiration else self.default_expiration
            self.cache[key] = (value, expire_at)
            print(f"[SET] Key '{key}' stored with TTL: {expiration or self.default_expiration or '∞'} seconds.")

    def get(self, key: str) -> Optional[Any]:
        """
        Retrieve a value from the cache.

        Args:
            key (str): The key to retrieve.
        Returns:
            Any: The value associated with the key, or None if the key does not exist or expired.
        """
        with self.lock:
            item = self.cache.get(key)
            if item:
                value, expire_at = item
                # Check if the key has expired
                if expire_at is None or expire_at > self._current_time():
                    print(f"[GET] Key '{key}' found with value: {value}")
                    return value
                else:
                    # Expired, so remove the key
                    print(f"[GET] Key '{key}' has expired.")
                    del self.cache[key]
        print(f"[GET] Key '{key}' not found.")
        return None

    def delete(self, key: str) -> bool:
        """
        Remove a key from the cache.

        Args:
            key (str): The key to delete.
        Returns:
            bool: True if the key existed and was removed, False otherwise.
        """
        with self.lock:
            if key in self.cache:
                del self.cache[key]
                print(f"[DELETE] Key '{key}' deleted.")
                return True
        print(f"[DELETE] Key '{key}' does not exist.")
        return False

    def exists(self, key: str) -> bool:
        """
        Check if a key exists in the cache and is not expired.

        Args:
            key (str): The key to check.
        Returns:
            bool: True if the key exists and is not expired, False otherwise.
        """
        with self.lock:
            if key in self.cache:
                _, expire_at = self.cache[key]
                if expire_at is None or expire_at > self._current_time():
                    print(f"[EXISTS] Key '{key}' exists.")
                    return True
                else:
                    # Key exists but expired
                    print(f"[EXISTS] Key '{key}' has expired.")
                    del self.cache[key]
        print(f"[EXISTS] Key '{key}' does not exist.")
        return False

    def flush(self):
        """
        Clears all keys from the cache.
        """
        with self.lock:
            self.cache.clear()
            print(f"[FLUSH] Cache cleared.")

    def keys(self):
        """
        Retrieve all keys currently in the cache.

        Returns:
            list: List of all active, non-expired keys in the cache.
        """
        with self.lock:
            valid_keys = []
            for key, (_, expire_at) in self.cache.items():
                if expire_at is None or expire_at > self._current_time():
                    valid_keys.append(key)
            print(f"[KEYS] Current active keys: {valid_keys}")
            return valid_keys

    def _garbage_collector(self):
        """
        Background thread for cleaning up expired keys.
        """
        while True:
            with self.lock:
                keys_to_delete = [
                    key for key, (_, expire_at) in self.cache.items()
                    if expire_at is not None and expire_at <= self._current_time()
                ]
                for key in keys_to_delete:
                    del self.cache[key]
            time.sleep(5)  # Run garbage collection every 5 seconds

if __name__ == "__main__":
    # Initialize the cache with a default expiration time of 10 seconds
    cache = PythonCache(default_expiration=10)

    # Set key-value pairs
    cache.set("name", "John Doe")  # Default expiration (10s)
    cache.set("framework", "Django", expiration=5)  # Custom expiration (5s)
    cache.set("language", "Python")  # Default expiration (10s)

    # Get values
    time.sleep(2)  # Wait 2 seconds
    cache.get("name")
    cache.get("framework")

    # Check existence
    cache.exists("language")
    time.sleep(6)  # Wait another 6 seconds
    cache.exists("framework")  # Should be expired by now

    # Delete a key
    cache.delete("name")

    # Get all active keys
    cache.keys()

    # Flush the cache
    cache.flush()
    cache.keys()