cache.py

import time
import threading

_cache = {}
_ttl = {}
_lock = threading.Lock()
DEFAULT_TTL = 60


def get(key: str):
    with _lock:
        entry = _cache.get(key)
        expiry = _ttl.get(key, 0)
        if entry and time.time() < expiry:
            return entry
        if entry:
            del _cache[key]
            del _ttl[key]
    return None


def set(key: str, data, ttl: int = DEFAULT_TTL):
    with _lock:
        _cache[key] = data
        _ttl[key] = time.time() + ttl


def invalidate(pattern: str = None):
    with _lock:
        if pattern:
            keys = [k for k in _cache if pattern in k]
            for k in keys:
                del _cache[k]
                del _ttl[k]
        else:
            _cache.clear()
            _ttl.clear()


def stats():
    with _lock:
        return {"items": len(_cache), "keys": list(_cache.keys())}