Cache Port

cache_port

Cache port for abstract caching operations.

Defines the CachePort contract 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

Abstract base class for caching operations.

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

Example

cache = MyCache[str, bytes]()
await cache.set("avatar", open("photo.png", "rb").read(), ttl=3600)
data = await cache.get("avatar")

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

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

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

    Example:
        ```python
        cache = MyCache[str, bytes]()
        await cache.set("avatar", open("photo.png", "rb").read(), ttl=3600)
        data = await cache.get("avatar")
        ```
    """

    @abstractmethod
    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.

        """
        ...

    @abstractmethod
    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.

        """

    @abstractmethod
    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.

        """

    @abstractmethod
    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.

        """
        ...

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

get(key: KeyType) -> ValueType | None abstractmethod 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

@abstractmethod
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 abstractmethod 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

@abstractmethod
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 abstractmethod 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

@abstractmethod
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 abstractmethod 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

@abstractmethod
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 abstractmethod async

Remove all entries from the cache.

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

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