Skip to main content

CachePort

What it is​

  • A small “port” module defining cache domain types and interfaces:
    • Exceptions for cache misses/expiry
    • A CachedData Pydantic model describing stored entries
    • Interfaces (ICacheAdapter, ICacheService) to be implemented by concrete cache adapters/services

Public API​

Exceptions​

  • CacheNotFoundError
    • Raised by implementations when a key is missing.
  • CacheExpiredError
    • Raised by implementations when an entry is expired (typically when TTL is enforced).

Enums​

  • DataType (Enum[str])
    • Supported data type labels for cached values:
      • TEXT, JSON, BINARY, PICKLE

Models​

  • CachedData (pydantic.BaseModel)
    • Represents a cached entry.
    • Fields:
      • key: str
      • data: Any
      • data_type: DataType
      • created_at: str (ISO timestamp; defaults to datetime.datetime.now().isoformat())

Interfaces​

  • ICacheAdapter

    • Low-level storage contract.
    • Methods (all raise NotImplementedError here):
      • get(key: str) -> CachedData
      • set(key: str, value: CachedData) -> None
      • delete(key: str) -> None
      • exists(key: str) -> bool

  • ICacheService

    • Higher-level cache service contract built on an adapter.
    • Attributes:
      • adapter: ICacheAdapter
    • Constructor:
      • __init__(adapter: ICacheAdapter) stores the adapter.
    • Methods (all raise NotImplementedError here):
      • exists(key: str) -> bool
      • get(key: str, ttl: datetime.timedelta | None = None) -> Any
      • set_text(key: str, value: str) -> None
      • set_json(key: str, value: dict) -> None
      • set_binary(key: str, value: bytes) -> None
      • set_pickle(key: str, value: Any) -> None

Configuration/Dependencies​

  • Dependencies:
    • pydantic (BaseModel, Field)
    • Standard library: datetime, enum, typing.Any
  • No runtime configuration is defined in this module.

Usage​

This file defines interfaces; you must implement them to use caching.

import datetime
from naas_abi_core.services.cache.CachePort import (
    ICacheAdapter, ICacheService, CachedData, DataType, CacheNotFoundError
)

class InMemoryAdapter(ICacheAdapter):
    def __init__(self):
        self._store = {}

    def get(self, key: str) -> CachedData:
        if key not in self._store:
            raise CacheNotFoundError(key)
        return self._store[key]

    def set(self, key: str, value: CachedData) -> None:
        self._store[key] = value

    def delete(self, key: str) -> None:
        self._store.pop(key, None)

    def exists(self, key: str) -> bool:
        return key in self._store

class SimpleCacheService(ICacheService):
    def exists(self, key: str) -> bool:
        return self.adapter.exists(key)

    def get(self, key: str, ttl: datetime.timedelta | None = None):
        return self.adapter.get(key).data

    def set_text(self, key: str, value: str) -> None:
        self.adapter.set(key, CachedData(key=key, data=value, data_type=DataType.TEXT))

adapter = InMemoryAdapter()
cache = SimpleCacheService(adapter)

cache.set_text("greeting", "hello")
print(cache.get("greeting"))

Caveats​

  • ICacheAdapter and ICacheService are not abc.ABC; they only raise NotImplementedError. Implementations must provide behavior.
  • TTL/expiry semantics are not implemented here; ttl is only part of the ICacheService.get interface.
  • CachedData.created_at is stored as a string timestamp (ISO format), not a datetime object.