Cache Port

cache_port

Cache port for abstract caching operations.

Defines the CachePort protocol that application code depends on,
decoupling caching from any specific implementation (in-memory, Redis, etc.).

Responsibilities

• Store and retrieve cached values by key.
• Check cache existence and clear entries.
• Support optional TTL (time-to-live) for entries.

Non-Responsibilities

• Eviction policies (LRU, LFU) — handled by infrastructure.
• Distributed locking or consistency guarantees.
• Serialization of cache values.

CachePort

Bases: OutboundPort[KeyType, ValueType], Protocol

Protocol for caching operations.

Any object with the async get, set, delete, exists,
and clear methods satisfies this protocol.

Class Type Parameters:

Name	Bound or Constraints	Description	Default
KeyType		The type of cache keys (typically str).	required
ValueType		The type of cached values.	required

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

class CachePort[KeyType, ValueType](
    OutboundPort[KeyType, ValueType],
    Protocol,
):
    """Protocol for caching operations.

    Any object with the async ``get``, ``set``, ``delete``, ``exists``,
    and ``clear`` methods satisfies this protocol.

    Type Parameters:
        KeyType: The type of cache keys (typically str).
        ValueType: The type of cached values.
    """

    async def get(self, key: KeyType) -> ValueType | None:
        """Retrieve a value from the cache.

        Args:
            key: The cache key.

        Returns:
            The cached value, or ``None`` if not found or expired.
        """
        ...

    async def set(
        self,
        key: KeyType,
        value: ValueType,
        ttl: float | None = None,
    ) -> None:
        """Store a value in the cache.

        Args:
            key: The cache key.
            value: The value to cache.
            ttl: Optional time-to-live in seconds. ``None`` means no expiration.
        """
        ...

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

        Args:
            key: The cache key. No-op if the key does not exist.
        """
        ...

    async def exists(self, key: KeyType) -> bool:
        """Check whether a key exists in the cache and has not expired.

        Args:
            key: The cache key.

        Returns:
            ``True`` if the key exists and is not expired, ``False`` otherwise.
        """
        ...

    async def clear(self) -> None:
        """Remove all entries from the cache."""
        ...

get(key: KeyType) -> ValueType | None async

Retrieve a value from the cache.

Parameters:

Name	Type	Description	Default
key	KeyType	The cache key.	required

Returns:

Type	Description
ValueType | None	The cached value, or None if not found or expired.

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

async def get(self, key: KeyType) -> ValueType | None:
    """Retrieve a value from the cache.

    Args:
        key: The cache key.

    Returns:
        The cached value, or ``None`` if not found or expired.
    """
    ...

set(key: KeyType, value: ValueType, ttl: float | None = None) -> None async

Store a value in the cache.

Parameters:

Name	Type	Description	Default
key	KeyType	The cache key.	required
value	ValueType	The value to cache.	required
ttl	float | None	Optional time-to-live in seconds. None means no expiration.	None

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

async def set(
    self,
    key: KeyType,
    value: ValueType,
    ttl: float | None = None,
) -> None:
    """Store a value in the cache.

    Args:
        key: The cache key.
        value: The value to cache.
        ttl: Optional time-to-live in seconds. ``None`` means no expiration.
    """
    ...

delete(key: KeyType) -> None async

Remove a value from the cache.

Parameters:

Name	Type	Description	Default
key	KeyType	The cache key. No-op if the key does not exist.	required

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

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

    Args:
        key: The cache key. No-op if the key does not exist.
    """
    ...

exists(key: KeyType) -> bool async

Check whether a key exists in the cache and has not expired.

Parameters:

Name	Type	Description	Default
key	KeyType	The cache key.	required

Returns:

Type	Description
bool	True if the key exists and is not expired, False otherwise.

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

async def exists(self, key: KeyType) -> bool:
    """Check whether a key exists in the cache and has not expired.

    Args:
        key: The cache key.

    Returns:
        ``True`` if the key exists and is not expired, ``False`` otherwise.
    """
    ...

clear() -> None async

Remove all entries from the cache.

Source code in src/forging_blocks/application/ports/outbound/cache_port.py

async def clear(self) -> None:
    """Remove all entries from the cache."""
    ...