cache

cache

Caching functionality for cocapi

CacheManager

CacheManager(default_ttl: int = 300)

Manages response caching with TTL support

Initialize cache manager

PARAMETER	DESCRIPTION
default_ttl	Default time-to-live in seconds (5 minutes) TYPE: int DEFAULT: 300

Source code in cocapi/cache.py

def __init__(self, default_ttl: int = 300):
    """
    Initialize cache manager

    Args:
        default_ttl: Default time-to-live in seconds (5 minutes)
    """
    self.cache: dict[str, CacheEntry] = {}
    self.default_ttl = default_ttl
    self._enabled = True
    self._stats = {
        "hits": 0,
        "misses": 0,
        "sets": 0,
        "evictions": 0,
    }

enable

enable() -> None

Enable caching

Source code in cocapi/cache.py

def enable(self) -> None:
    """Enable caching"""
    self._enabled = True

disable

disable() -> None

Disable caching

Source code in cocapi/cache.py

def disable(self) -> None:
    """Disable caching"""
    self._enabled = False

is_enabled

is_enabled() -> bool

Check if caching is enabled

Source code in cocapi/cache.py

def is_enabled(self) -> bool:
    """Check if caching is enabled"""
    return self._enabled

get

get(
    url: str, params: dict[str, Any] | None = None
) -> dict[str, Any] | None

Get cached response if available and not expired

PARAMETER	DESCRIPTION
url	Request URL TYPE: str
params	Request parameters TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
dict[str, Any] | None	Cached response data or None if not found/expired

Source code in cocapi/cache.py

def get(
    self, url: str, params: dict[str, Any] | None = None
) -> dict[str, Any] | None:
    """
    Get cached response if available and not expired

    Args:
        url: Request URL
        params: Request parameters

    Returns:
        Cached response data or None if not found/expired
    """
    if not self._enabled:
        return None

    cache_key = get_cache_key(url, params)
    current_time = time.time()

    if cache_key in self.cache:
        entry = self.cache[cache_key]

        if not entry.is_expired(current_time):
            self._stats["hits"] += 1
            return entry.data
        else:
            # Remove expired entry
            del self.cache[cache_key]
            self._stats["evictions"] += 1

    self._stats["misses"] += 1
    return None

set

set(
    url: str,
    params: dict[str, Any] | None,
    data: dict[str, Any],
    ttl: int | None = None,
) -> None

Cache response data

PARAMETER	DESCRIPTION
url	Request URL TYPE: str
params	Request parameters TYPE: dict[str, Any] | None
data	Response data to cache TYPE: dict[str, Any]
ttl	Time-to-live in seconds (uses default if not specified) TYPE: int | None DEFAULT: None

Source code in cocapi/cache.py

def set(
    self,
    url: str,
    params: dict[str, Any] | None,
    data: dict[str, Any],
    ttl: int | None = None,
) -> None:
    """
    Cache response data

    Args:
        url: Request URL
        params: Request parameters
        data: Response data to cache
        ttl: Time-to-live in seconds (uses default if not specified)
    """
    if not self._enabled:
        return

    # Don't cache error responses
    if data.get("result") == "error":
        return

    cache_key = get_cache_key(url, params)
    ttl = ttl or self.default_ttl

    self.cache[cache_key] = CacheEntry(
        data=data.copy(), timestamp=time.time(), ttl=ttl
    )

    self._stats["sets"] += 1

invalidate

invalidate(
    url: str, params: dict[str, Any] | None = None
) -> bool

Invalidate specific cached entry

PARAMETER	DESCRIPTION
url	Request URL TYPE: str
params	Request parameters TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
bool	True if entry was found and removed, False otherwise

Source code in cocapi/cache.py

def invalidate(self, url: str, params: dict[str, Any] | None = None) -> bool:
    """
    Invalidate specific cached entry

    Args:
        url: Request URL
        params: Request parameters

    Returns:
        True if entry was found and removed, False otherwise
    """
    cache_key = get_cache_key(url, params)

    if cache_key in self.cache:
        del self.cache[cache_key]
        self._stats["evictions"] += 1
        return True

    return False

clear

clear() -> int

Clear all cached entries

RETURNS	DESCRIPTION
int	Number of entries cleared

Source code in cocapi/cache.py

def clear(self) -> int:
    """
    Clear all cached entries

    Returns:
        Number of entries cleared
    """
    count = len(self.cache)
    self.cache.clear()
    self._stats["evictions"] += count
    return count

cleanup_expired

cleanup_expired() -> int

Remove expired entries

RETURNS	DESCRIPTION
int	Number of expired entries removed

Source code in cocapi/cache.py

def cleanup_expired(self) -> int:
    """
    Remove expired entries

    Returns:
        Number of expired entries removed
    """
    current_time = time.time()
    expired_keys = [
        key for key, entry in self.cache.items() if entry.is_expired(current_time)
    ]

    for key in expired_keys:
        del self.cache[key]

    self._stats["evictions"] += len(expired_keys)
    return len(expired_keys)

get_stats

get_stats() -> dict[str, Any]

Get cache statistics

Source code in cocapi/cache.py

def get_stats(self) -> dict[str, Any]:
    """Get cache statistics"""
    total_requests = self._stats["hits"] + self._stats["misses"]
    hit_rate = (
        (self._stats["hits"] / total_requests * 100) if total_requests > 0 else 0
    )

    # Count expired entries
    current_time = time.time()
    expired_count = sum(
        1 for entry in self.cache.values() if entry.is_expired(current_time)
    )

    return {
        "enabled": self._enabled,
        "total_entries": len(self.cache),
        "expired_entries": expired_count,
        "valid_entries": len(self.cache) - expired_count,
        "default_ttl": self.default_ttl,
        "hit_rate": round(hit_rate, 2),
        "stats": self._stats.copy(),
    }

get_cache_info

get_cache_info() -> dict[str, Any]

Get detailed cache information

Source code in cocapi/cache.py

def get_cache_info(self) -> dict[str, Any]:
    """Get detailed cache information"""
    current_time = time.time()
    entries = []

    for key, entry in self.cache.items():
        entries.append(
            {
                "key": key,
                "size_estimate": len(str(entry.data)),
                "ttl": entry.ttl,
                "age": round(current_time - entry.timestamp, 2),
                "expires_in": round(
                    (entry.timestamp + entry.ttl) - current_time, 2
                ),
                "is_expired": entry.is_expired(current_time),
            }
        )

    # Sort by expiration time (soonest to expire first)
    from typing import cast

    entries.sort(key=lambda x: cast(float, x["expires_in"]))

    return {
        "entries": entries,
        "memory_usage_estimate": sum(e["size_estimate"] for e in entries),
    }

set_default_ttl

set_default_ttl(ttl: int) -> None

Set default TTL for new cache entries

Source code in cocapi/cache.py

def set_default_ttl(self, ttl: int) -> None:
    """Set default TTL for new cache entries"""
    self.default_ttl = ttl

get_entry_info

get_entry_info(
    url: str, params: dict[str, Any] | None = None
) -> dict[str, Any] | None

Get information about a specific cache entry

Source code in cocapi/cache.py

def get_entry_info(
    self, url: str, params: dict[str, Any] | None = None
) -> dict[str, Any] | None:
    """Get information about a specific cache entry"""
    cache_key = get_cache_key(url, params)

    if cache_key not in self.cache:
        return None

    entry = self.cache[cache_key]
    current_time = time.time()

    return {
        "key": cache_key,
        "ttl": entry.ttl,
        "age": round(current_time - entry.timestamp, 2),
        "expires_in": round((entry.timestamp + entry.ttl) - current_time, 2),
        "is_expired": entry.is_expired(current_time),
        "size_estimate": len(str(entry.data)),
        "cached_at": entry.timestamp,
    }

extend_ttl

extend_ttl(
    url: str,
    params: dict[str, Any] | None = None,
    additional_seconds: int = 300,
) -> bool

Extend TTL for a specific cache entry

PARAMETER	DESCRIPTION
url	Request URL TYPE: str
params	Request parameters TYPE: dict[str, Any] | None DEFAULT: None
additional_seconds	Seconds to add to current TTL TYPE: int DEFAULT: 300

RETURNS	DESCRIPTION
bool	True if entry was found and updated, False otherwise

Source code in cocapi/cache.py

def extend_ttl(
    self,
    url: str,
    params: dict[str, Any] | None = None,
    additional_seconds: int = 300,
) -> bool:
    """
    Extend TTL for a specific cache entry

    Args:
        url: Request URL
        params: Request parameters
        additional_seconds: Seconds to add to current TTL

    Returns:
        True if entry was found and updated, False otherwise
    """
    cache_key = get_cache_key(url, params)

    if cache_key in self.cache:
        entry = self.cache[cache_key]
        entry.ttl += additional_seconds
        return True

    return False

touch

touch(
    url: str, params: dict[str, Any] | None = None
) -> bool

Reset timestamp for cache entry (extends its life without changing TTL)

PARAMETER	DESCRIPTION
url	Request URL TYPE: str
params	Request parameters TYPE: dict[str, Any] | None DEFAULT: None

RETURNS	DESCRIPTION
bool	True if entry was found and touched, False otherwise

Source code in cocapi/cache.py

def touch(self, url: str, params: dict[str, Any] | None = None) -> bool:
    """
    Reset timestamp for cache entry (extends its life without changing TTL)

    Args:
        url: Request URL
        params: Request parameters

    Returns:
        True if entry was found and touched, False otherwise
    """
    cache_key = get_cache_key(url, params)

    if cache_key in self.cache:
        entry = self.cache[cache_key]
        entry.timestamp = time.time()
        return True

    return False