Skip to content

Types

Specialized type implementations for advanced use cases.

xpytools.xtype.TTLSet.TTLSet

Python
TTLSet(ttl: int = 600, maxsize: int = 512, sweep_interval: int = 50)

Thread-safe set with time-to-live (TTL) expiration and auto-sweeping.

This behaves like a lightweight, memory-safe cache for tracking transient keys (such as job IDs or request IDs) that should automatically expire after a set duration.

It uses an OrderedDict to maintain insertion order, and an internal RLock for thread-safe access. Expired keys are purged automatically every few insertions, and old entries are evicted when the maxsize limit is reached.

Parameters:

Name Type Description Default

ttl

int

Time-to-live in seconds for each entry.

 600

maxsize

int

Maximum number of cached entries before oldest entries are evicted.

 512

sweep_interval

int

Number of insertions between automatic sweeps.

 50
Example
Python
from xpytools.types import TTLSet

seen = TTLSet(ttl=10, maxsize=100)

seen.add("file_1")
"file_1" in seen  # True

import time; time.sleep(11)
"file_1" in seen  # False — expired automatically
Notes
  • Uses monotonic time internally (safe against system clock changes).
  • Thread-safe (uses threading.RLock).
  • Designed for transient ID tracking or deduplication.

Methods:

Name Description
add

Add a key to the set, resetting its expiration timestamp.
clear

Remove all entries from the cache immediately.
sweep

Manually remove all expired entries.
Source code in xpytools/xtype/TTLSet.py
Python
def __init__(self, ttl: int = 600, maxsize: int = 512, sweep_interval: int = 50):
    self._ttl = ttl
    self._maxsize = maxsize
    self._sweep_interval = sweep_interval
    self._insert_count = 0
    self._cache: OrderedDict[str, float] = OrderedDict()
    self._lock = threading.RLock()

add

Python
add(key: str) -> None

Add a key to the set, resetting its expiration timestamp.

If the key already exists, its TTL is refreshed. A cleanup sweep is triggered automatically after every sweep_interval insertions.

Parameters:

Name Type Description Default
key
str

The key to store.

 required
Source code in xpytools/xtype/TTLSet.py
Python
def add(self, key: str) -> None:
    """
    Add a key to the set, resetting its expiration timestamp.

    If the key already exists, its TTL is refreshed. A cleanup sweep is
    triggered automatically after every `sweep_interval` insertions.

    Parameters
    ----------
    key : str
        The key to store.
    """
    now = monotonic()
    with self._lock:
        self._cache[key] = now + self._ttl
        self._cache.move_to_end(key)
        self._insert_count += 1

        # Automatic sweep every N insertions
        if self._insert_count >= self._sweep_interval:
            self._insert_count = 0
            self._sweep_locked(now)

        # Enforce maxsize cap
        if len(self._cache) > self._maxsize:
            self._cache.popitem(last=False)

clear

Python
clear() -> None

Remove all entries from the cache immediately.

Source code in xpytools/xtype/TTLSet.py
Python
def clear(self) -> None:
    """Remove all entries from the cache immediately."""
    with self._lock:
        self._cache.clear()

sweep

Python
sweep() -> None

Manually remove all expired entries.

Source code in xpytools/xtype/TTLSet.py
Python
def sweep(self) -> None:
    """Manually remove all expired entries."""
    with self._lock:
        self._sweep_locked()

xpytools.xtype.UUIDLike.UUIDLike module-attribute

Python
UUIDLike = cast(type[str], _vUUIDFactory())