cache

mlflow_secrets_auth.cache

TTL cache implementation for secrets.

Provides a lightweight, thread-safe cache with monotonic-clock-based TTLs and a simple decorator (cached_fetch) to memoize zero-argument callables.

Design goals

• Monotonic time to avoid issues when the wall clock changes.
• Thread safety via RLock.
• No caching of failures: exceptions from the wrapped callable return None and are not stored.
• Global cache instance for convenience, with helpers to clear and inspect size.

TTLCache()

Thread-safe TTL cache (monotonic-clock based).

Initialize an empty TTL cache with thread safety.

Source code in src/mlflow_secrets_auth/cache.py

def __init__(self) -> None:
    """Initialize an empty TTL cache with thread safety."""
    self._cache: dict[str, tuple[Any, float]] = {}
    self._lock = threading.RLock()

clear()

Clear all cached items.

Source code in src/mlflow_secrets_auth/cache.py

def clear(self) -> None:
    """Clear all cached items."""
    with self._lock:
        self._cache.clear()

delete(key)

Remove a key from the cache.

Parameters:

Name	Type	Description	Default
key	str	Cache key to remove.	required

Source code in src/mlflow_secrets_auth/cache.py

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

    Args:
        key: Cache key to remove.

    """
    with self._lock:
        self._cache.pop(key, None)

get(key)

Get a value from the cache if present and not expired.

Parameters:

Name	Type	Description	Default
key	str	Cache key.	required

Returns:

Type	Description
Any | None	The cached value if present and valid, otherwise None.

Source code in src/mlflow_secrets_auth/cache.py

def get(self, key: str) -> Any | None:
    """Get a value from the cache if present and not expired.

    Args:
        key: Cache key.

    Returns:
        The cached value if present and valid, otherwise None.

    """
    with self._lock:
        entry = self._cache.get(key)
        if entry is None:
            return None
        value, expiry = entry
        if self._now() > expiry:
            self._cache.pop(key, None)
            return None
        return value

invalidate_prefix(prefix)

Remove all keys starting with a prefix.

Useful for provider-wide invalidation using e.g. f"{provider_name}:".

Parameters:

Name	Type	Description	Default
prefix	str	Prefix to match.	required

Source code in src/mlflow_secrets_auth/cache.py

def invalidate_prefix(self, prefix: str) -> None:
    """Remove all keys starting with a prefix.

    Useful for provider-wide invalidation using e.g. ``f"{provider_name}:"``.

    Args:
        prefix: Prefix to match.

    """
    with self._lock:
        to_delete = [k for k in self._cache if k.startswith(prefix)]
        for k in to_delete:
            self._cache.pop(k, None)

set(key, value, ttl_seconds)

Set a value in the cache with a TTL.

Non-positive or sub-minimum TTLs are treated as "no caching" (the key is removed). TTLs larger than MAX_TTL_SECONDS are capped.

Parameters:

Name	Type	Description	Default
key	str	Cache key.	required
value	Any	Value to store.	required
ttl_seconds	float	Time-to-live in seconds.	required

Source code in src/mlflow_secrets_auth/cache.py

def set(self, key: str, value: Any, ttl_seconds: float) -> None:
    """Set a value in the cache with a TTL.

    Non-positive or sub-minimum TTLs are treated as "no caching"
    (the key is removed). TTLs larger than MAX_TTL_SECONDS are capped.

    Args:
        key: Cache key.
        value: Value to store.
        ttl_seconds: Time-to-live in seconds.

    """
    with self._lock:
        if ttl_seconds < MIN_TTL_SECONDS:
            self._cache.pop(key, None)
            return
        ttl = min(float(ttl_seconds), float(MAX_TTL_SECONDS))
        self._cache[key] = (value, self._now() + ttl)

size()

Return the current cache size, pruning expired entries first.

Returns:

Type	Description
int	Number of live (non-expired) entries.

Source code in src/mlflow_secrets_auth/cache.py

def size(self) -> int:
    """Return the current cache size, pruning expired entries first.

    Returns:
        Number of live (non-expired) entries.

    """
    with self._lock:
        now = self._now()
        to_delete = [k for k, (_, exp) in self._cache.items() if now > exp]
        for k in to_delete:
            self._cache.pop(k, None)
        return len(self._cache)

cached_fetch(cache_key, ttl_seconds=DEFAULT_TTL_SECONDS)

Cache a zero-argument function's result with a TTL.

Exceptions raised by the wrapped function are swallowed and result in None, which is not cached. Successful non-None results are cached.

Parameters:

Name	Type	Description	Default
cache_key	str	Unique cache key for the function result.	required
ttl_seconds	int	Time-to-live for the cached value.	DEFAULT_TTL_SECONDS

Returns:

Type	Description
Callable[[Callable[[], T]], Callable[[], T | None]]	A decorator that wraps a Callable[[], T] and returns Callable[[], Optional[T]].

Source code in src/mlflow_secrets_auth/cache.py

def cached_fetch(cache_key: str, ttl_seconds: int = DEFAULT_TTL_SECONDS) -> Callable[[Callable[[], T]], Callable[[], T | None]]:
    """Cache a zero-argument function's result with a TTL.

    Exceptions raised by the wrapped function are swallowed and result in `None`,
    which is not cached. Successful non-None results are cached.

    Args:
        cache_key: Unique cache key for the function result.
        ttl_seconds: Time-to-live for the cached value.

    Returns:
        A decorator that wraps a `Callable[[], T]` and returns `Callable[[], Optional[T]]`.

    """

    def decorator(fetch_func: Callable[[], T]) -> Callable[[], T | None]:
        def wrapper() -> T | None:
            cached_value = _global_cache.get(cache_key)
            if cached_value is not None:
                # Typing note: caller-provided T is preserved by construction.
                return cached_value  # type: ignore[return-value]

            try:
                value = fetch_func()
            except Exception:
                # Do not cache failures; return None to the caller.
                return None

            if value is not None:
                _global_cache.set(cache_key, value, ttl_seconds)
            return value

        return wrapper

    return decorator

clear_cache()

Clear the global cache instance.

Source code in src/mlflow_secrets_auth/cache.py

def clear_cache() -> None:
    """Clear the global cache instance."""
    _global_cache.clear()

delete_cache_key(key)

Remove a single cache entry by key.

Parameters:

Name	Type	Description	Default
key	str	Cache key to remove.	required

Source code in src/mlflow_secrets_auth/cache.py

def delete_cache_key(key: str) -> None:
    """Remove a single cache entry by key.

    Args:
        key: Cache key to remove.

    """
    _global_cache.delete(key)

get_cache_size()

Get the current size of the global cache (after pruning).

Source code in src/mlflow_secrets_auth/cache.py

def get_cache_size() -> int:
    """Get the current size of the global cache (after pruning)."""
    return _global_cache.size()