Generic Python TTL cache that uses threads

ttl_cache.py

import threading
import time

from typing import Callable, TypeVar, Generic, Any, Dict, Tuple, List

T = TypeVar("T")


class TTLCache(Generic[T]):
    def __init__(
        self,
        ttl: float,
        on_expire: Callable[[T], Any] | None = None,
    ):
        self.ttl = ttl
        self.on_expire = on_expire
        self.cache: Dict[str, Tuple[T, float]] = {}

        self.lock = threading.Lock()
        self.start_cleanup_thread()

    def set(self, key: str, value: T):
        with self.lock:
            self.cache[key] = (value, time.time() + self.ttl)

    def get(self, key: str):
        with self.lock:
            if key in self.cache:
                value, expire_time = self.cache[key]
                if time.time() < expire_time:
                    # Reset the TTL for the item
                    self.cache[key] = (value, time.time() + self.ttl)
                    return value

    def __len__(self):
        with self.lock:
            return len(self.cache)

    def keys(self):
        with self.lock:
            return [key for (key, _) in self.cache.keys()]

    def delete(self, key: str):
        value: T | None = None

        with self.lock:
            if key in self.cache:
                value, _ = self.cache[key]
                del self.cache[key]

        if self.on_expire and value:
            self.on_expire(value)

    def cleanup(self):
        values: List[T] = []

        with self.lock:
            for key in list(self.cache.keys()):
                _, expire_time = self.cache[key]
                if time.time() >= expire_time:
                    values.append(self.cache[key][0])
                    del self.cache[key]

        if self.on_expire:
            for value in values:
                self.on_expire(value)

    def clear(self):
        values: List[T] = []

        with self.lock:
            for key in list(self.cache.keys()):
                values.append(self.cache[key][0])
                del self.cache[key]

        if self.on_expire:
            for value in values:
                self.on_expire(value)

    def start_cleanup_thread(self):
        def cleanup_thread():
            while True:
                time.sleep(self.ttl / 10)
                self.cleanup()

        thread = threading.Thread(target=cleanup_thread, daemon=True)
        thread.start()