Cache

provide.foundation.transport.cache

TODO: Add module docstring.

Classes

TransportCache

Transport cache with automatic health-based eviction.

Tracks transport health and automatically evicts transports that exceed the failure threshold. This prevents cascading failures from unhealthy transports.

Functions

clear

clear() -> dict[str, Any]

Clear all cached transports.

Returns:

Type	Description
dict[str, Any]	Dictionary of evicted transports for cleanup

Source code in provide/foundation/transport/cache.py

def clear(self) -> dict[str, Any]:
    """Clear all cached transports.

    Returns:
        Dictionary of evicted transports for cleanup
    """
    log.debug("🧹 Clearing transport cache", count=len(self._transports))
    transports = dict(self._transports)
    self._transports.clear()
    self._health.clear()
    self._evicted.clear()
    return transports

evict

evict(scheme: str) -> None

Evict transport from cache.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme to evict	required

Source code in provide/foundation/transport/cache.py

def evict(self, scheme: str) -> None:
    """Evict transport from cache.

    Args:
        scheme: Transport scheme to evict
    """
    if scheme in self._transports:
        log.info("🗑️ Evicting transport", scheme=scheme)
        # Don't await disconnect - caller handles cleanup
        self._transports.pop(scheme, None)
        self._health.pop(scheme, None)
        self._evicted.add(scheme)

get_health

get_health(scheme: str) -> TransportHealth | None

Get health status for scheme.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme	required

Returns:

Type	Description
TransportHealth | None	TransportHealth or None if not tracked

Source code in provide/foundation/transport/cache.py

def get_health(self, scheme: str) -> TransportHealth | None:
    """Get health status for scheme.

    Args:
        scheme: Transport scheme

    Returns:
        TransportHealth or None if not tracked
    """
    return self._health.get(scheme)

get_or_create async

get_or_create(
    scheme: str, factory: Callable[[str], Any]
) -> Any

Get or create transport for scheme.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme (e.g., "http", "https")	required
factory	Callable[[str], Any]	Factory function to create new transport	required

Returns:

Type	Description
Any	Transport instance

Raises:

Type	Description
TransportCacheEvictedError	If transport was evicted due to failures

Source code in provide/foundation/transport/cache.py

async def get_or_create(
    self,
    scheme: str,
    factory: Callable[[str], Any],
) -> Any:
    """Get or create transport for scheme.

    Args:
        scheme: Transport scheme (e.g., "http", "https")
        factory: Factory function to create new transport

    Returns:
        Transport instance

    Raises:
        TransportCacheEvictedError: If transport was evicted due to failures
    """
    # Check if transport was evicted
    if scheme in self._evicted:
        raise TransportCacheEvictedError(
            f"Transport '{scheme}' was evicted due to {self.failure_threshold} consecutive failures. "
            f"Clear cache or create new client to retry.",
            scheme=scheme,
            consecutive_failures=self.failure_threshold,
        )

    # Get or create transport
    if scheme not in self._transports:
        transport = factory(scheme)
        await transport.connect()
        self._transports[scheme] = transport
        self._health[scheme] = TransportHealth()

    return self._transports[scheme]

is_evicted

is_evicted(scheme: str) -> bool

Check if transport was evicted.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme	required

Returns:

Type	Description
bool	True if transport was evicted

Source code in provide/foundation/transport/cache.py

def is_evicted(self, scheme: str) -> bool:
    """Check if transport was evicted.

    Args:
        scheme: Transport scheme

    Returns:
        True if transport was evicted
    """
    return scheme in self._evicted

mark_failure

mark_failure(scheme: str, error: Exception) -> None

Mark a failed request for scheme.

Automatically evicts transport if failure threshold is exceeded.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme	required
error	Exception	Error that occurred	required

Source code in provide/foundation/transport/cache.py

def mark_failure(self, scheme: str, error: Exception) -> None:
    """Mark a failed request for scheme.

    Automatically evicts transport if failure threshold is exceeded.

    Args:
        scheme: Transport scheme
        error: Error that occurred
    """
    if scheme not in self._health:
        # Transport not in cache, nothing to track
        return

    health = self._health[scheme]
    health.record_failure()

    log.warning(
        "⚠️ Transport request failed",
        scheme=scheme,
        consecutive_failures=health.consecutive_failures,
        failure_rate=f"{health.failure_rate:.2%}",
        error=str(error),
    )

    # Check if we should evict (threshold=0 disables eviction)
    if self.failure_threshold > 0 and health.consecutive_failures >= self.failure_threshold:
        self.evict(scheme)
        log.error(
            "🚫 Transport evicted due to consecutive failures",
            scheme=scheme,
            consecutive_failures=health.consecutive_failures,
            total_failures=health.total_failures,
            total_requests=health.total_requests,
        )

mark_success

mark_success(scheme: str) -> None

Mark a successful request for scheme.

Parameters:

Name	Type	Description	Default
scheme	str	Transport scheme	required

Source code in provide/foundation/transport/cache.py

def mark_success(self, scheme: str) -> None:
    """Mark a successful request for scheme.

    Args:
        scheme: Transport scheme
    """
    if scheme in self._health:
        self._health[scheme].record_success()

TransportHealth

Health tracking for a transport instance.

Attributes

failure_rate property

failure_rate: float

Calculate failure rate.

Functions

record_failure

record_failure() -> None

Record a failed request.

Source code in provide/foundation/transport/cache.py

def record_failure(self) -> None:
    """Record a failed request."""
    self.total_requests += 1
    self.total_failures += 1
    self.consecutive_failures += 1

record_success

record_success() -> None

Record a successful request.

Source code in provide/foundation/transport/cache.py

def record_success(self) -> None:
    """Record a successful request."""
    self.total_requests += 1
    self.consecutive_failures = 0

Functions