Cache

TenantCache

Python

TenantCache(max_size: int = 1000, ttl: int = 60)

In-process LRU cache with per-entry TTL for tenant objects.

Parameters:

Name	Type	Description	Default
max_size	int	Maximum number of tenant entries to hold (LRU eviction when full). Default: 1000.	1000
ttl	int	Seconds before an entry is considered stale. Default: 60.	60

Example::

Text Only

cache = TenantCache(max_size=500, ttl=30)

cache.set(tenant)
hit = cache.get(tenant.id)           # by ID
hit = cache.get_by_identifier("acme-corp")  # by slug

cache.invalidate(tenant.id)
cache.clear()

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def __init__(self, max_size: int = 1000, ttl: int = 60) -> None:
    if max_size < 1:
        msg = "max_size must be >= 1"
        raise ValueError(msg)
    if ttl < 1:
        msg = "ttl must be >= 1 second"
        raise ValueError(msg)

    self._max_size = max_size
    self._ttl = ttl
    # Both dicts maintain insertion order; move_to_end keeps MRU at back.
    self._by_id: OrderedDict[str, _Entry] = OrderedDict()
    self._id_by_ident: dict[str, str] = {}  # identifier → tenant_id

    # Telemetry counters — monotonically increasing, never reset.
    self._hits: int = 0
    self._misses: int = 0

    # guard all dict mutations so the cache is safe under
    # concurrent asyncio tasks (including anyio/trio backends).
    # Created lazily via _get_lock() to avoid needing a running event
    # loop at __init__ time (e.g. synchronous test setup).
    self._lock: asyncio.Lock | None = None

    logger.debug("TenantCache initialised max_size=%d ttl=%ds", max_size, ttl)

get

Python

get(tenant_id: str) -> Tenant | None

Return the cached tenant for tenant_id, or None on miss/expiry.

On a cache hit, the entry is promoted to the MRU position.

Parameters:

Name	Type	Description	Default
tenant_id	str	Opaque tenant primary key.	required

Returns:

Type	Description
Tenant | None	Cached Tenant, or None on miss or expiry.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def get(self, tenant_id: str) -> Tenant | None:
    """Return the cached tenant for *tenant_id*, or ``None`` on miss/expiry.

    On a cache hit, the entry is promoted to the MRU position.

    Args:
        tenant_id: Opaque tenant primary key.

    Returns:
        Cached ``Tenant``, or ``None`` on miss or expiry.
    """
    entry = self._by_id.get(tenant_id)
    if entry is None:
        self._misses += 1
        return None
    if time.monotonic() > entry.expires_at:
        self._evict(tenant_id)
        self._misses += 1
        return None
    self._by_id.move_to_end(tenant_id)
    self._hits += 1
    return entry.tenant

get_by_identifier

Python

get_by_identifier(identifier: str) -> Tenant | None

Return the cached tenant for identifier, or None on miss/expiry.

Parameters:

Name	Type	Description	Default
identifier	str	Human-readable tenant slug.	required

Returns:

Type	Description
Tenant | None	Cached Tenant, or None on miss or expiry.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def get_by_identifier(self, identifier: str) -> Tenant | None:
    """Return the cached tenant for *identifier*, or ``None`` on miss/expiry.

    Args:
        identifier: Human-readable tenant slug.

    Returns:
        Cached ``Tenant``, or ``None`` on miss or expiry.
    """
    tenant_id = self._id_by_ident.get(identifier)
    if tenant_id is None:
        return None
    return self.get(tenant_id)

set

Python

set(tenant: Tenant) -> None

Insert or refresh tenant in the cache.

When the cache is at capacity, the LRU entry is evicted before inserting the new entry.

Thread/task safety: all mutations are protected by _get_lock(). Call await cache.aset(tenant) from async contexts to benefit from the lock; set() is kept synchronous for backwards-compatibility but should only be called before any concurrent tasks access the cache.

Parameters:

Name	Type	Description	Default
tenant	Tenant	The tenant to cache. Both ID and identifier keys are written atomically.	required

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def set(self, tenant: Tenant) -> None:
    """Insert or refresh *tenant* in the cache.

    When the cache is at capacity, the LRU entry is evicted before
    inserting the new entry.

    Thread/task safety: all mutations are protected by ``_get_lock()``.
    Call ``await cache.aset(tenant)`` from async contexts to benefit from
    the lock; ``set()`` is kept synchronous for backwards-compatibility
    but should only be called before any concurrent tasks access the cache.

    Args:
        tenant: The tenant to cache.  Both ID and identifier keys are
            written atomically.
    """
    self._set_unsafe(tenant)

aset async

Python

aset(tenant: Tenant) -> None

Async variant of set() — acquires the mutex before writing.

Use this in async code paths (middleware, resolvers, cache proxies) to guarantee safety under concurrent task scheduling.

Parameters:

Name	Type	Description	Default
tenant	Tenant	The tenant to cache.	required

Source code in src/fastapi_tenancy/cache/tenant_cache.py

async def aset(self, tenant: Tenant) -> None:
    """Async variant of ``set()`` — acquires the mutex before writing.

    Use this in async code paths (middleware, resolvers, cache proxies)
    to guarantee safety under concurrent task scheduling.

    Args:
        tenant: The tenant to cache.
    """
    async with self._get_lock():
        self._set_unsafe(tenant)

invalidate

Python

invalidate(tenant_id: str) -> bool

Remove the cache entry for tenant_id.

Parameters:

Name	Type	Description	Default
tenant_id	str	Opaque tenant primary key.	required

Returns:

Type	Description
bool	True when an entry was found and removed; False on miss.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def invalidate(self, tenant_id: str) -> bool:
    """Remove the cache entry for *tenant_id*.

    Args:
        tenant_id: Opaque tenant primary key.

    Returns:
        ``True`` when an entry was found and removed; ``False`` on miss.
    """
    return self._evict(tenant_id)

invalidate_by_identifier

Python

invalidate_by_identifier(identifier: str) -> bool

Remove the cache entry for identifier.

Parameters:

Name	Type	Description	Default
identifier	str	Human-readable tenant slug.	required

Returns:

Type	Description
bool	True when an entry was removed; False on miss.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def invalidate_by_identifier(self, identifier: str) -> bool:
    """Remove the cache entry for *identifier*.

    Args:
        identifier: Human-readable tenant slug.

    Returns:
        ``True`` when an entry was removed; ``False`` on miss.
    """
    tenant_id = self._id_by_ident.get(identifier)
    if tenant_id is None:
        return False
    return self._evict(tenant_id)

clear

Python

clear() -> int

Evict all entries and reset both indices.

Returns:

Type	Description
int	Number of entries evicted.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def clear(self) -> int:
    """Evict all entries and reset both indices.

    Returns:
        Number of entries evicted.
    """
    count = len(self._by_id)
    self._by_id.clear()
    self._id_by_ident.clear()
    logger.debug("TenantCache cleared (%d entries evicted)", count)
    return count

size

Python

size() -> int

Return the number of currently cached entries.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def size(self) -> int:
    """Return the number of currently cached entries."""
    return len(self._by_id)

stats

Python

stats() -> dict[str, int]

Return a snapshot of cache state for monitoring.

Returns:

Type	Description
dict[str, int]	Dictionary with: - size: current number of entries. - max_size: configured maximum. - ttl: configured TTL in seconds. - hits: cumulative cache hit count since creation. - misses: cumulative cache miss count since creation. - hit_rate_pct: integer hit-rate percentage (0-100), or 0 when no lookups have occurred yet.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def stats(self) -> dict[str, int]:
    """Return a snapshot of cache state for monitoring.

    Returns:
        Dictionary with:
            - ``size``: current number of entries.
            - ``max_size``: configured maximum.
            - ``ttl``: configured TTL in seconds.
            - ``hits``: cumulative cache hit count since creation.
            - ``misses``: cumulative cache miss count since creation.
            - ``hit_rate_pct``: integer hit-rate percentage (0-100),
              or 0 when no lookups have occurred yet.
    """
    total = self._hits + self._misses
    hit_rate = int(self._hits * 100 / total) if total > 0 else 0
    return {
        "size": len(self._by_id),
        "max_size": self._max_size,
        "ttl": self._ttl,
        "hits": self._hits,
        "misses": self._misses,
        "hit_rate_pct": hit_rate,
    }

purge_expired

Python

purge_expired() -> int

Eagerly remove all expired entries.

Not required — entries are lazily evicted on access — but useful to call periodically in low-traffic applications to reclaim memory.

Returns:

Type	Description
int	Number of entries evicted.

Source code in src/fastapi_tenancy/cache/tenant_cache.py

def purge_expired(self) -> int:
    """Eagerly remove all expired entries.

    Not required — entries are lazily evicted on access — but useful
    to call periodically in low-traffic applications to reclaim memory.

    Returns:
        Number of entries evicted.
    """
    now = time.monotonic()
    expired_keys = [k for k, entry in self._by_id.items() if now > entry.expires_at]
    for key in expired_keys:
        self._evict(key)
    return len(expired_keys)