Caching

Classes for caching loaded configurations.

CachedLoader

Loader wrapper that caches results.

CachedLoader

CachedLoader(
    loader: Any,
    cache: CacheBackend[
        DaprAgentConfig | WorkflowDefinition
    ]
    | None = None,
    *,
    default_ttl: float | None = 300.0,
)

A caching wrapper around DaprAgentSpecLoader.

Provides transparent caching for loaded specifications to avoid reparsing the same content multiple times.

Example

from dapr_agents_oas_adapter import DaprAgentSpecLoader
from dapr_agents_oas_adapter.cache import CachedLoader, InMemoryCache

# Create a cached loader with 5-minute TTL
cache = InMemoryCache(default_ttl=300)
loader = CachedLoader(DaprAgentSpecLoader(), cache)

# First call parses the JSON
config1 = loader.load_json(json_string)

# Second call returns a cached result
config2 = loader.load_json(json_string)

Initialize the cached loader.

Parameters:

Name	Type	Description	Default
loader	Any	The underlying DaprAgentSpecLoader instance	required
cache	CacheBackend[DaprAgentConfig | WorkflowDefinition] | None	Cache backend to use (creates InMemoryCache if None)	None
default_ttl	float | None	Default TTL for cached entries in seconds	300.0

Source code in src/dapr_agents_oas_adapter/cache.py

def __init__(
    self,
    loader: Any,  # DaprAgentSpecLoader, but avoid circular import
    cache: CacheBackend[DaprAgentConfig | WorkflowDefinition] | None = None,
    *,
    default_ttl: float | None = 300.0,  # 5 minutes default
) -> None:
    """Initialize the cached loader.

    Args:
        loader: The underlying DaprAgentSpecLoader instance
        cache: Cache backend to use (creates InMemoryCache if None)
        default_ttl: Default TTL for cached entries in seconds
    """
    self._loader = loader
    self._cache: CacheBackend[DaprAgentConfig | WorkflowDefinition] = (
        cache if cache is not None else InMemoryCache(default_ttl=default_ttl)
    )
    self._default_ttl = default_ttl
    self._logger = get_logger()
    self._stats = CacheStats()

Attributes

stats property

stats: CacheStats

Get cache statistics.

Functions

load_yaml

load_yaml(
    yaml_content: str, *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from an YAML string with caching.

Parameters:

Name	Type	Description	Default
yaml_content	str	YAML string containing the OAS specification	required
use_cache	bool	Whether to use caching (default: True)	True

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Source code in src/dapr_agents_oas_adapter/cache.py

def load_yaml(
    self, yaml_content: str, *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from an YAML string with caching.

    Args:
        yaml_content: YAML string containing the OAS specification
        use_cache: Whether to use caching (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows
    """
    if not use_cache:
        return self._loader.load_yaml(yaml_content)

    cache_key = _compute_cache_key(yaml_content, prefix="yaml")
    return self._get_or_load(cache_key, lambda: self._loader.load_yaml(yaml_content))

load_json

load_json(
    json_content: str, *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from a JSON string with caching.

Parameters:

Name	Type	Description	Default
json_content	str	JSON string containing the OAS specification	required
use_cache	bool	Whether to use caching (default: True)	True

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Source code in src/dapr_agents_oas_adapter/cache.py

def load_json(
    self, json_content: str, *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from a JSON string with caching.

    Args:
        json_content: JSON string containing the OAS specification
        use_cache: Whether to use caching (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows
    """
    if not use_cache:
        return self._loader.load_json(json_content)

    cache_key = _compute_cache_key(json_content, prefix="json")
    return self._get_or_load(cache_key, lambda: self._loader.load_json(json_content))

load_dict

load_dict(
    spec_dict: dict[str, Any], *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition

Load an OAS specification from a dictionary with caching.

Parameters:

Name	Type	Description	Default
spec_dict	dict[str, Any]	Dictionary containing the OAS specification	required
use_cache	bool	Whether to use caching (default: True)	True

Returns:

Type	Description
DaprAgentConfig | WorkflowDefinition	DaprAgentConfig for Agent components, WorkflowDefinition for Flows

Source code in src/dapr_agents_oas_adapter/cache.py

def load_dict(
    self, spec_dict: dict[str, Any], *, use_cache: bool = True
) -> DaprAgentConfig | WorkflowDefinition:
    """Load an OAS specification from a dictionary with caching.

    Args:
        spec_dict: Dictionary containing the OAS specification
        use_cache: Whether to use caching (default: True)

    Returns:
        DaprAgentConfig for Agent components, WorkflowDefinition for Flows
    """
    if not use_cache:
        return self._loader.load_dict(spec_dict)

    cache_key = _compute_cache_key(spec_dict, prefix="dict")
    return self._get_or_load(cache_key, lambda: self._loader.load_dict(spec_dict))

InMemoryCache

Thread-safe in-memory cache with TTL support.

InMemoryCache

InMemoryCache(
    default_ttl: float | None = None,
    max_size: int | None = None,
)

Bases: CacheBackend[T]

Thread-safe in-memory cache with TTL support.

Initialize the cache.

Parameters:

Name	Type	Description	Default
default_ttl	float | None	Default time-to-live in seconds (None = no expiration)	None
max_size	int | None	Maximum number of entries (None = unlimited)	None

Source code in src/dapr_agents_oas_adapter/cache.py

def __init__(self, default_ttl: float | None = None, max_size: int | None = None) -> None:
    """Initialize the cache.

    Args:
        default_ttl: Default time-to-live in seconds (None = no expiration)
        max_size: Maximum number of entries (None = unlimited)
    """
    self._cache: dict[str, CacheEntry[T]] = {}
    self._lock = threading.RLock()
    self._default_ttl = default_ttl
    self._max_size = max_size
    self._logger = get_logger()

Functions

get

get(key: str) -> T | None

Get a value from the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

def get(self, key: str) -> T | None:
    """Get a value from the cache."""
    with self._lock:
        entry = self._cache.get(key)
        if entry is None:
            self._logger.debug("cache_miss", extra={"key": key})
            return None

        if entry.is_expired():
            self._logger.debug("cache_expired", extra={"key": key})
            del self._cache[key]
            return None

        self._logger.debug("cache_hit", extra={"key": key})
        return entry.value

set

set(key: str, value: T, ttl: float | None = None) -> None

Set a value in the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

def set(self, key: str, value: T, ttl: float | None = None) -> None:
    """Set a value in the cache."""
    with self._lock:
        # Use provided TTL or default
        actual_ttl = ttl if ttl is not None else self._default_ttl

        # Calculate expiration time
        now = time.time()
        expires_at = now + actual_ttl if actual_ttl is not None else None

        # Enforce max size by removing the oldest entries
        if (
            self._max_size is not None
            and len(self._cache) >= self._max_size
            and key not in self._cache
        ):
            self._evict_oldest()

        self._cache[key] = CacheEntry(value=value, created_at=now, expires_at=expires_at)
        self._logger.debug("cache_set", extra={"key": key, "ttl": actual_ttl})

delete

delete(key: str) -> bool

Delete a value from the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

def delete(self, key: str) -> bool:
    """Delete a value from the cache."""
    with self._lock:
        if key in self._cache:
            del self._cache[key]
            self._logger.debug("cache_deleted", extra={"key": key})
            return True
        return False

clear

clear() -> None

Clear all entries from the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

def clear(self) -> None:
    """Clear all entries from the cache."""
    with self._lock:
        count = len(self._cache)
        self._cache.clear()
        self._logger.debug("cache_cleared", extra={"entries_removed": count})

CacheBackend

Abstract base class for cache implementations.

CacheBackend

Bases: ABC

Abstract cache backend interface.

Functions

get abstractmethod

get(key: str) -> T | None

Get a value from the cache.

Parameters:

Name	Type	Description	Default
key	str	The cache key	required

Returns:

Type	Description
T | None	The cached value or None if not found/expired

Source code in src/dapr_agents_oas_adapter/cache.py

@abstractmethod
def get(self, key: str) -> T | None:
    """Get a value from the cache.

    Args:
        key: The cache key

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

set abstractmethod

set(key: str, value: T, ttl: float | None = None) -> None

Set a value in the cache.

Parameters:

Name	Type	Description	Default
key	str	The cache key	required
value	T	The value to cache	required
ttl	float | None	Time-to-live in seconds (None = no expiration)	None

Source code in src/dapr_agents_oas_adapter/cache.py

@abstractmethod
def set(self, key: str, value: T, ttl: float | None = None) -> None:
    """Set a value in the cache.

    Args:
        key: The cache key
        value: The value to cache
        ttl: Time-to-live in seconds (None = no expiration)
    """
    ...

delete abstractmethod

delete(key: str) -> bool

Delete a value from the cache.

Parameters:

Name	Type	Description	Default
key	str	The cache key	required

Returns:

Type	Description
bool	True if the key was deleted, False if not found

Source code in src/dapr_agents_oas_adapter/cache.py

@abstractmethod
def delete(self, key: str) -> bool:
    """Delete a value from the cache.

    Args:
        key: The cache key

    Returns:
        True if the key was deleted, False if not found
    """
    ...

clear abstractmethod

clear() -> None

Clear all entries from the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

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

size abstractmethod

size() -> int

Get the number of entries in the cache.

Source code in src/dapr_agents_oas_adapter/cache.py

@abstractmethod
def size(self) -> int:
    """Get the number of entries in the cache."""
    ...

CacheStats

Cache statistics tracking.

CacheStats dataclass

CacheStats(hits: int = 0, misses: int = 0)

Statistics for cache performance.

Attributes

hits class-attribute instance-attribute

hits: int = 0

misses class-attribute instance-attribute

misses: int = 0

Usage Examples

Basic Caching

from dapr_agents_oas_adapter import (
    CachedLoader,
    DaprAgentSpecLoader,
    InMemoryCache
)

# Create cache
cache = InMemoryCache(
    max_size=100,
    ttl_seconds=300
)

# Create cached loader
loader = CachedLoader(
    loader=DaprAgentSpecLoader(),
    cache=cache
)

# Load (cache miss)
config = loader.load_yaml(yaml_content)

# Load again (cache hit)
config = loader.load_yaml(yaml_content)

# Check stats
print(f"Hits: {loader.stats.hits}")
print(f"Misses: {loader.stats.misses}")

Custom Cache Backend

from dapr_agents_oas_adapter import CacheBackend
from typing import TypeVar

T = TypeVar("T")

class MyCache(CacheBackend[T]):
    def get(self, key: str) -> T | None:
        # Custom get implementation
        pass

    def set(self, key: str, value: T) -> None:
        # Custom set implementation
        pass

    def delete(self, key: str) -> bool:
        # Custom delete implementation
        pass

    def clear(self) -> None:
        # Custom clear implementation
        pass

Related

• DaprAgentSpecLoader - Base loader
• Loading Guide - Usage guide