Inmemory Cache¶

def is_expired(self) -> bool:
    """Check if this entry has expired."""
    if self.expire_at is None:
        return False
    return time.time() > self.expire_at

def get_ttl(self) -> int | None:
    """Get remaining TTL in seconds."""
    if self.expire_at is None:
        return None
    remaining = int(self.expire_at - time.time())
    return max(0, remaining)

async def set_negative(self, key: str, /) -> None:
    """Store cache miss marker to prevent cache penetration."""
    async with self._lock:
        try:
            await self._evict_if_needed()
            entry = TTLEntry(NEGATIVE, self.negative_ttl)
            self._cache[key] = entry
            self.logger.debug(f"Set negative cache for key: {key}")
        except Exception as e:
            self.logger.warning(f"Failed to set negative cache for key {key}: {e}")

async def keys(self) -> list[str]:
    """Return a list of cache keys."""
    result = []
    async for key in self.iter_keys():
        result.append(key)
    return result

async def get(self, key: str, default: VT | T | None = None, /) -> VT | T | None:
    """Get value from cache, return default if not found."""
    result = await self.get_item(key)
    if isinstance(result, Empty):
        return default
    return result  # type: ignore

async def setdefault(self, key: str, default: VT | None = None, /) -> VT | None:
    """Get value or set and return default if key doesn't exist."""
    async with self._lock:
        try:
            await self._cleanup_expired()
            entry = self._cache.get(key)

            if entry is not None and not entry.is_expired():
                if isinstance(entry.value, CacheMiss):
                    self.logger.debug(f"Key {key} in negative cache")
                    return default
                self.logger.debug(f"Key {key} exists, returning cached value")
                return entry.value  # type: ignore

            # Key doesn't exist or is expired
            if entry is not None and entry.is_expired():
                del self._cache[key]

            if default is not None:
                await self._evict_if_needed()
                new_entry = TTLEntry(default, self.default_ttl)
                self._cache[key] = new_entry
                self.logger.debug(f"Set default value for key: {key}")
                return default

            await self._evict_if_needed()
            neg_entry = TTLEntry(NEGATIVE, self.negative_ttl)
            self._cache[key] = neg_entry
            self.logger.debug(f"Set negative cache for key: {key}")
            return None

        except Exception as e:
            self.logger.error(f"Error in setdefault for key {key}: {e}")
            return default

async def clear(self) -> None:
    """Clear all cache entries with the configured prefix."""
    async with self._lock:
        try:
            keys_to_delete = [key for key in self._cache if self.key_builder.validate(key)]

            for key in keys_to_delete:
                del self._cache[key]

            self.logger.info(f"Cleared {len(keys_to_delete)} cache entries")
        except Exception as e:
            self.logger.error(f"Failed to clear cache: {e}")

async def pop(self, key: str, default: VT | T | None = None, /) -> VT | T | None:
    """Remove and return value, or return default if not found."""
    async with self._lock:
        try:
            await self._cleanup_expired()
            entry = self._cache.get(key)

            if entry is not None and not entry.is_expired():
                del self._cache[key]
                if isinstance(entry.value, CacheMiss):
                    return default
                self.logger.debug(f"Successfully popped key: {key}")
                return entry.value  # type: ignore

            self.logger.debug(f"Key not found for pop operation: {key}")
            return default
        except Exception as e:
            self.logger.error(f"Error when popping key {key}: {e}")
            return default

async def popitem(self) -> tuple[str, VT]:
    """Remove and return an arbitrary (key, value) pair."""
    async with self._lock:
        try:
            await self._cleanup_expired()

            for key in list(self._cache.keys()):
                if not self.key_builder.validate(key):
                    continue

                entry = self._cache.get(key)
                if entry is not None and not entry.is_expired():
                    del self._cache[key]
                    if not isinstance(entry.value, CacheMiss):
                        self.logger.debug(f"Successfully popped item: {key}")
                        return key, entry.value

            self.logger.debug("No items to pop")
            raise KeyError("popitem(): cache is empty")
        except KeyError:
            raise
        except Exception as e:
            self.logger.error(f"Error during popitem: {e}")
            raise KeyError(f"popitem() failed: {e}") from e

async def set(self, key: str, value: VT, /) -> None:
    """Set a key-value pair (alias for set_item)."""
    await self.set_item(key, value)

async def setx(self, key: str, value: VT, /, ttl: int | None = None) -> None:
    """Set a key-value pair with optional TTL."""
    async with self._lock:
        try:
            await self._evict_if_needed()
            entry = TTLEntry(value, ttl)
            self._cache[key] = entry

            if ttl is not None:
                self.logger.debug(f"Set key {key} with TTL {ttl}")
            else:
                self.logger.debug(f"Set key {key} without TTL")
        except Exception as e:
            self.logger.error(f"Failed to setx for key {key}: {e}")

async def ttl(self, key: str, /) -> int | None:
    """Get TTL for a key."""
    async with self._lock:
        try:
            entry = self._cache.get(key)

            if entry is None:
                self.logger.debug(f"Key {key} does not exist")
                return None

            if entry.is_expired():
                del self._cache[key]
                self.logger.debug(f"Key {key} has expired")
                return None

            ttl_value = entry.get_ttl()
            if ttl_value is None:
                self.logger.debug(f"Key {key} exists without expiration")
            else:
                self.logger.debug(f"Key {key} TTL: {ttl_value}")

            return ttl_value
        except Exception as e:
            self.logger.error(f"Error when getting TTL for key {key}: {e}")
            return None

async def expire(self, key: str, /, ttl: int | None = None) -> None:
    """Set or remove expiration for a key."""
    async with self._lock:
        try:
            entry = self._cache.get(key)

            if entry is None:
                self.logger.warning(f"Cannot set expiration for non-existent key: {key}")
                return

            if entry.is_expired():
                del self._cache[key]
                self.logger.warning(f"Key {key} has already expired")
                return

            # Create new entry with updated TTL
            new_entry = TTLEntry(entry.value, ttl)
            self._cache[key] = new_entry

            if ttl is not None:
                self.logger.debug(f"Set expiration for key {key}: {ttl} seconds")
            else:
                self.logger.debug(f"Removed expiration for key: {key}")
        except Exception as e:
            self.logger.error(f"Error when setting expiration for key {key}: {e}")

async def incr(self, key: str, /, amount: int = 1) -> int:
    """Increment a key's value."""
    async with self._lock:
        try:
            await self._cleanup_expired()
            entry = self._cache.get(key)

            if entry is None or entry.is_expired():
                # Initialize to amount if key doesn't exist
                new_value = amount
                await self._evict_if_needed()
                new_entry = TTLEntry(new_value, self.default_ttl)
                self._cache[key] = new_entry
                self.logger.debug(f"Initialized and incremented key {key} to {new_value}")
                return new_value

            if not isinstance(entry.value, int):
                self.logger.error(f"Cannot increment non-integer value for key {key}")
                raise ValueError(f"Value at key {key} is not an integer")

            new_value = entry.value + amount
            # Preserve existing TTL
            existing_ttl = entry.get_ttl()
            new_entry = TTLEntry(new_value, existing_ttl)
            self._cache[key] = new_entry

            self.logger.debug(f"Incremented key {key} by {amount}, result: {new_value}")
            return new_value
        except Exception as e:
            self.logger.error(f"Error when incrementing key {key}: {e}")
            return 0

async def decr(self, key: str, /, amount: int = 1) -> int:
    """Decrement a key's value."""
    return await self.incr(key, -amount)

async def values(self) -> list[VT]:
    """Return all cache values."""
    async with self._lock:
        try:
            await self._cleanup_expired()
            values = []
            for key, entry in self._cache.items():
                if not self.key_builder.validate(key):
                    continue
                if entry.is_expired():
                    continue
                if isinstance(entry.value, CacheMiss):
                    continue
                values.append(entry.value)
            return values
        except Exception as e:
            self.logger.error(f"Error when getting values: {e}")
            return []

async def items(self) -> list[tuple[str, VT]]:
    """Return all cache items as (key, value) pairs."""
    async with self._lock:
        try:
            await self._cleanup_expired()
            items = []
            for key, entry in self._cache.items():
                if not self.key_builder.validate(key):
                    continue
                if entry.is_expired():
                    continue
                if isinstance(entry.value, CacheMiss):
                    continue
                items.append((key, entry.value))
            return items
        except Exception as e:
            self.logger.error(f"Error when getting items: {e}")
            return []

async def init(self) -> None:
    """Initialize cache resources if needed."""
    self.logger.info("InmemoryCache initialized")

async def close(self) -> None:
    """Close cache and cleanup resources."""
    async with self._lock:
        try:
            self._cache.clear()
            self.logger.info("Cachetools cache closed and cleared")
        except Exception as e:
            self.logger.warning(f"Error while closing cache: {e}")