classes

CacheInstance dataclass

Bases: CacheInstanceBase

Class to control a Diskcache Cache instance.

Parameters:

Name	Type	Description	Default
cache_dir	str | Path	Directory path where cache.db will be stored.	field(default=CACHE_DIR)
index	bool	Controls creation of a tag index in the cache instance.	field(default=True)
cache	Cache	A diskcache.Cache object. When the class is instantiated, a Cache will be created.	field(default=None)
cache_timeout(int)		Default key expiration (in seconds).	required

Source code in src/red_utils/ext/diskcache_utils/classes.py

@dataclass
class CacheInstance(CacheInstanceBase):
    """Class to control a Diskcache Cache instance.

    Params:
        cache_dir (str|Path): Directory path where cache.db will be stored.
        index (bool): Controls creation of a tag index in the cache instance.
        cache (diskcache.Cache): A diskcache.Cache object. When the class is instantiated, a Cache will be created.
        cache_timeout(int): Default key expiration (in seconds).
    """

    def check_key_exists(self, key: valid_key_types = None) -> bool:
        """Check if a key exists in a cache.

        Params:
            key (str): The cache key to search for

        Returns:
            (bool): `True` if cache key found
            (bool): `False` if cache key not found

        """
        ## Key validation
        validate_key(key=key)
        validate_cache(cache=self.cache)

        ## Check if key exists in cache
        if key in self.cache:
            return True
        else:
            return False

    def set_val(
        self,
        key: valid_key_types = None,
        val: valid_val_types = None,
        expire: int = None,
        read: bool = False,
        tag: str = None,
        retry: bool = False,
    ) -> None:
        """Set a key value pair in the cache.

        Params:
            key (str): The key to store the value under in the cache
            val (str): The value to store in the cache
            expire (int): Time (in seconds) before value expires
            read (bool): If `True`, read value as a file-like object
            tag (str): Applies a tag to the cached value
            retry (bool): If `True`, retry setting cache key if first attempt fails
        """
        validate_key(key)
        validate_val(val)
        validate_expire(expire, none_ok=True)
        validate_read(read, none_ok=True)
        validate_tag(tag=tag, none_ok=True)
        validate_retry(retry=retry, none_ok=True)
        validate_cache(cache=self.cache)

        try:
            with self.cache as ref:
                ref.set(
                    key=key, value=val, expire=expire, read=read, tag=tag, retry=retry
                )

        except Exception as exc:
            raise Exception(
                f"Unhandled exception setting key/value pair for key: [{key}]. Details: {exc}"
            )

    def get_val(self, key: valid_key_types = None, tags: list[str] = None):
        """Search for a key in a given cache.

        Pass a diskcache.Cache object for cache, and a key (and optionally a list of tags).
        Function will search the cache and return a value if found, or a structured
        error dict describing the lack of key.

        Params:
            key (str): The key to search the cache for
            tags (list[str]): List of tags to search the cache for
        """
        validate_key(key)
        validate_cache(self.cache)
        validate_tags(tags)

        try:
            if check_cache_key_exists(key=key, cache=self.cache):
                try:
                    with self.cache as ref:
                        _val = ref.get(key=key)

                        return _val

                except Exception as exc:
                    raise Exception(
                        f"Unhandled exception retrieving value of key [{key}]. Details: {exc}"
                    )

            else:
                return {
                    "error": "Key not found in cache",
                    "details": {"key": key, "cache_dir": self.cache.directory},
                }

        except Exception as exc:
            return {
                "error": "Error searching for key in cache",
                "details": {"exception": exc},
            }

    def set_expire(
        self, key: valid_key_types = None, expire: int = None
    ) -> Union[dict[str, str], None]:
        """Set an expiration timeout (in seconds).

        Params:
            key (str): Name of the key to set expiration on. Must already exist in the cache.
            expire (int): Time (in seconds) to wait before expiring cached value.
        """
        validate_key(key)
        validate_cache(self.cache)
        validate_expire(expire)

        if not check_cache_key_exists(key=key, cache=self.cache):
            return {
                "warning": f"Cache item with key [{key}] does not exist in cache at {self.cache.directory}/"
            }

        try:
            with self.cache as ref:
                ref.touch(key, expire=expire)

        except Exception as exc:
            raise Exception(
                f"Unhandled exception setting expiration of {expire} on key [{key}] in cache at {self.cache.directory}/. Details: {exc}"
            )

    def delete_val(self, key: valid_key_types = None, tag: str = None) -> tuple:
        """Delete a cached value.

        If a tag is provided, only keys that also have that tag will be deleted.

        Params:
            key (str|int): Name of key in cache.
        """
        validate_key(key)
        validate_cache(self.cache)
        validate_tag(tag)

        try:
            with self.cache as ref:
                _delete = ref.pop(key=key, tag=tag)

                return _delete

        except Exception as exc:
            raise Exception(
                f"Unhandled exception deleting key {key} from cache at {self.cache.directory}/. Details: {exc}"
            )

    def get_cache_size(self) -> dict[str, int]:
        """Get a dict describing the size of the cache, in bytes.

        Returns:
            (dict): A Python `dict` with keys: 'unit', 'size'. Example return object:
                `{'unit': 'bytes', 'size': 36864}`

        """
        validate_cache(cache=self.cache)

        try:
            cache_size: int = self.cache.volume()

            return {"unit": "bytes", "size": cache_size}

        except Exception as exc:
            raise Exception(f"Unhandled exception getting cache size. Details: {exc}")

    def check_cache(self) -> list[warnings.WarningMessage]:
        """Run checks on Cache instance.

        Returns:
            (list[warning.WarningMessage]): A list of Diskcache `WarningMessage` objects.

        """
        validate_cache(cache=self.cache)

        try:
            warnings = self.cache.check()

            return warnings

        except Exception as exc:
            raise Exception(
                f"Unhandled exception checking cache for warnings. Details: {exc}"
            )

check_cache()

Run checks on Cache instance.

Returns:

Type	Description
list[WarningMessage]	A list of Diskcache WarningMessage objects.

Source code in src/red_utils/ext/diskcache_utils/classes.py

def check_cache(self) -> list[warnings.WarningMessage]:
    """Run checks on Cache instance.

    Returns:
        (list[warning.WarningMessage]): A list of Diskcache `WarningMessage` objects.

    """
    validate_cache(cache=self.cache)

    try:
        warnings = self.cache.check()

        return warnings

    except Exception as exc:
        raise Exception(
            f"Unhandled exception checking cache for warnings. Details: {exc}"
        )

check_key_exists(key=None)

Check if a key exists in a cache.

Parameters:

Name	Type	Description	Default
key	str	The cache key to search for	None

Returns:

Type	Description
bool	True if cache key found
bool	False if cache key not found

Source code in src/red_utils/ext/diskcache_utils/classes.py

def check_key_exists(self, key: valid_key_types = None) -> bool:
    """Check if a key exists in a cache.

    Params:
        key (str): The cache key to search for

    Returns:
        (bool): `True` if cache key found
        (bool): `False` if cache key not found

    """
    ## Key validation
    validate_key(key=key)
    validate_cache(cache=self.cache)

    ## Check if key exists in cache
    if key in self.cache:
        return True
    else:
        return False

delete_val(key=None, tag=None)

Delete a cached value.

If a tag is provided, only keys that also have that tag will be deleted.

Parameters:

Name	Type	Description	Default
key	str | int	Name of key in cache.	None

Source code in src/red_utils/ext/diskcache_utils/classes.py

def delete_val(self, key: valid_key_types = None, tag: str = None) -> tuple:
    """Delete a cached value.

    If a tag is provided, only keys that also have that tag will be deleted.

    Params:
        key (str|int): Name of key in cache.
    """
    validate_key(key)
    validate_cache(self.cache)
    validate_tag(tag)

    try:
        with self.cache as ref:
            _delete = ref.pop(key=key, tag=tag)

            return _delete

    except Exception as exc:
        raise Exception(
            f"Unhandled exception deleting key {key} from cache at {self.cache.directory}/. Details: {exc}"
        )

get_cache_size()

Get a dict describing the size of the cache, in bytes.

Returns:

Type	Description
dict	A Python dict with keys: 'unit', 'size'. Example return object: {'unit': 'bytes', 'size': 36864}

Source code in src/red_utils/ext/diskcache_utils/classes.py

def get_cache_size(self) -> dict[str, int]:
    """Get a dict describing the size of the cache, in bytes.

    Returns:
        (dict): A Python `dict` with keys: 'unit', 'size'. Example return object:
            `{'unit': 'bytes', 'size': 36864}`

    """
    validate_cache(cache=self.cache)

    try:
        cache_size: int = self.cache.volume()

        return {"unit": "bytes", "size": cache_size}

    except Exception as exc:
        raise Exception(f"Unhandled exception getting cache size. Details: {exc}")

get_val(key=None, tags=None)

Search for a key in a given cache.

Pass a diskcache.Cache object for cache, and a key (and optionally a list of tags). Function will search the cache and return a value if found, or a structured error dict describing the lack of key.

Parameters:

Name	Type	Description	Default
key	str	The key to search the cache for	None
tags	list[str]	List of tags to search the cache for	None

Source code in src/red_utils/ext/diskcache_utils/classes.py

def get_val(self, key: valid_key_types = None, tags: list[str] = None):
    """Search for a key in a given cache.

    Pass a diskcache.Cache object for cache, and a key (and optionally a list of tags).
    Function will search the cache and return a value if found, or a structured
    error dict describing the lack of key.

    Params:
        key (str): The key to search the cache for
        tags (list[str]): List of tags to search the cache for
    """
    validate_key(key)
    validate_cache(self.cache)
    validate_tags(tags)

    try:
        if check_cache_key_exists(key=key, cache=self.cache):
            try:
                with self.cache as ref:
                    _val = ref.get(key=key)

                    return _val

            except Exception as exc:
                raise Exception(
                    f"Unhandled exception retrieving value of key [{key}]. Details: {exc}"
                )

        else:
            return {
                "error": "Key not found in cache",
                "details": {"key": key, "cache_dir": self.cache.directory},
            }

    except Exception as exc:
        return {
            "error": "Error searching for key in cache",
            "details": {"exception": exc},
        }

set_expire(key=None, expire=None)

Set an expiration timeout (in seconds).

Parameters:

Name	Type	Description	Default
key	str	Name of the key to set expiration on. Must already exist in the cache.	None
expire	int	Time (in seconds) to wait before expiring cached value.	None

Source code in src/red_utils/ext/diskcache_utils/classes.py

def set_expire(
    self, key: valid_key_types = None, expire: int = None
) -> Union[dict[str, str], None]:
    """Set an expiration timeout (in seconds).

    Params:
        key (str): Name of the key to set expiration on. Must already exist in the cache.
        expire (int): Time (in seconds) to wait before expiring cached value.
    """
    validate_key(key)
    validate_cache(self.cache)
    validate_expire(expire)

    if not check_cache_key_exists(key=key, cache=self.cache):
        return {
            "warning": f"Cache item with key [{key}] does not exist in cache at {self.cache.directory}/"
        }

    try:
        with self.cache as ref:
            ref.touch(key, expire=expire)

    except Exception as exc:
        raise Exception(
            f"Unhandled exception setting expiration of {expire} on key [{key}] in cache at {self.cache.directory}/. Details: {exc}"
        )

set_val(key=None, val=None, expire=None, read=False, tag=None, retry=False)

Set a key value pair in the cache.

Parameters:

Name	Type	Description	Default
key	str	The key to store the value under in the cache	None
val	str	The value to store in the cache	None
expire	int	Time (in seconds) before value expires	None
read	bool	If True, read value as a file-like object	False
tag	str	Applies a tag to the cached value	None
retry	bool	If True, retry setting cache key if first attempt fails	False

Source code in src/red_utils/ext/diskcache_utils/classes.py

def set_val(
    self,
    key: valid_key_types = None,
    val: valid_val_types = None,
    expire: int = None,
    read: bool = False,
    tag: str = None,
    retry: bool = False,
) -> None:
    """Set a key value pair in the cache.

    Params:
        key (str): The key to store the value under in the cache
        val (str): The value to store in the cache
        expire (int): Time (in seconds) before value expires
        read (bool): If `True`, read value as a file-like object
        tag (str): Applies a tag to the cached value
        retry (bool): If `True`, retry setting cache key if first attempt fails
    """
    validate_key(key)
    validate_val(val)
    validate_expire(expire, none_ok=True)
    validate_read(read, none_ok=True)
    validate_tag(tag=tag, none_ok=True)
    validate_retry(retry=retry, none_ok=True)
    validate_cache(cache=self.cache)

    try:
        with self.cache as ref:
            ref.set(
                key=key, value=val, expire=expire, read=read, tag=tag, retry=retry
            )

    except Exception as exc:
        raise Exception(
            f"Unhandled exception setting key/value pair for key: [{key}]. Details: {exc}"
        )

CacheInstanceBase dataclass

Bases: DictMixin

Compose a Diskcache Cache from class parameters.

Parameters:

Name	Type	Description	Default
cache_dir	str | Path	The directory where the cache database should be created	field(default=CACHE_DIR)
index	bool	If True, a database index will be created	field(default=True)
cache	Cache	A diskcache.Cache instance	field(default=None)
cache_timeout	int	Default expiration time (in seconds)	field(default_factory=default_timeout)

Source code in src/red_utils/ext/diskcache_utils/classes.py

@dataclass
class CacheInstanceBase(DictMixin):
    """Compose a Diskcache Cache from class parameters.

    Params:
        cache_dir (str|Path): The directory where the cache database should be created
        index (bool): If `True`, a database index will be created
        cache (diskcache.Cache): A `diskcache.Cache` instance
        cache_timeout (int): Default expiration time (in seconds)
    """

    cache_dir: Union[str, Path] | None = field(default=CACHE_DIR)
    index: bool = field(default=True)
    cache: Cache | None = field(default=None)
    cache_timeout: int | None = field(default_factory=default_timeout)

    def __post_init__(self):
        if isinstance(self.cache_dir, str):
            if self.cache_dir == ".":
                self.cache_dir = Path().absolute()
            else:
                self.cache_dir = Path(self.cache_dir)

    def exists(self) -> bool:
        return Path(f"{self.cache_dir}/cache.db").exists()

    @property
    def cache_path(self) -> Path:
        return Path(f"{self.cache_dir}/cache.db")

    @property
    def cache_conf_dict(self) -> dict[str, Any]:
        _config = {
            "directory": self.cache_dir,
            "timeout": self.cache_timeout,
        }

        return _config

    def init(self) -> Cache:
        """Initialize a Diskcache Cache from class parameters.

        Sets the self.cache parameter to the initialized Cache,
        and also returns Cache directly.

        Returns:
            (diskcache.Cache): An initialized `DiskCache.Cache` object

        """
        try:
            cache = Cache(self.cache_dir, timeout=self.cache_timeout)
            self.cache = cache

            self.manage_cache_tag_index("create")

            return cache

        except Exception as exc:
            raise Exception(f"Unhandled exception creating cache. Details: {exc}")

    def manage_cache_tag_index(self, operation: str = "create") -> None:
        """Create or delete a cache index.

        Params:
            operation (str): The operation to perform on the cache's tag index.
                Options: ["create", "delete"]
        """
        valid_operations: list[str] = ["create", "delete"]

        validate_cache(cache=self.cache)

        if not operation:
            raise Exception(f"Operation cannot be None.")

        try:
            match operation:
                case "create":
                    if self.cache.tag_index == 0:
                        self.cache.create_tag_index()
                    else:
                        pass

                case "delete":
                    if self.cache.tag_index == 1:
                        self.cache.drop_tag_index()
                    else:
                        pass

                case _:
                    raise ValueError(
                        f"Invalid operation: {operation}. Must be one of {valid_operations}"
                    )

        except Exception as exc:
            raise Exception(
                f"Unhandled exception configuring tag_index. Details: {exc}"
            )

    def clear(self) -> bool:
        """Clear the entire cache.

        Returns:
            (bool): `True` if clearing cache successful
            (bool): `False` if clearing the cache not successful

        """
        validate_cache(self.cache)

        try:
            with self.cache as ref:
                ref.clear()

                return True

        except Exception as exc:
            raise Exception(
                f"Unhandled exception clearing cache at {self.cache.directory}. Details: {exc}"
            )

clear()

Clear the entire cache.

Returns:

Type	Description
bool	True if clearing cache successful
bool	False if clearing the cache not successful

Source code in src/red_utils/ext/diskcache_utils/classes.py

def clear(self) -> bool:
    """Clear the entire cache.

    Returns:
        (bool): `True` if clearing cache successful
        (bool): `False` if clearing the cache not successful

    """
    validate_cache(self.cache)

    try:
        with self.cache as ref:
            ref.clear()

            return True

    except Exception as exc:
        raise Exception(
            f"Unhandled exception clearing cache at {self.cache.directory}. Details: {exc}"
        )

init()

Initialize a Diskcache Cache from class parameters.

Sets the self.cache parameter to the initialized Cache, and also returns Cache directly.

Returns:

Type	Description
Cache	An initialized DiskCache.Cache object

Source code in src/red_utils/ext/diskcache_utils/classes.py

def init(self) -> Cache:
    """Initialize a Diskcache Cache from class parameters.

    Sets the self.cache parameter to the initialized Cache,
    and also returns Cache directly.

    Returns:
        (diskcache.Cache): An initialized `DiskCache.Cache` object

    """
    try:
        cache = Cache(self.cache_dir, timeout=self.cache_timeout)
        self.cache = cache

        self.manage_cache_tag_index("create")

        return cache

    except Exception as exc:
        raise Exception(f"Unhandled exception creating cache. Details: {exc}")

manage_cache_tag_index(operation='create')

Create or delete a cache index.

Parameters:

Name	Type	Description	Default
operation	str	The operation to perform on the cache's tag index. Options: ["create", "delete"]	'create'

Source code in src/red_utils/ext/diskcache_utils/classes.py

def manage_cache_tag_index(self, operation: str = "create") -> None:
    """Create or delete a cache index.

    Params:
        operation (str): The operation to perform on the cache's tag index.
            Options: ["create", "delete"]
    """
    valid_operations: list[str] = ["create", "delete"]

    validate_cache(cache=self.cache)

    if not operation:
        raise Exception(f"Operation cannot be None.")

    try:
        match operation:
            case "create":
                if self.cache.tag_index == 0:
                    self.cache.create_tag_index()
                else:
                    pass

            case "delete":
                if self.cache.tag_index == 1:
                    self.cache.drop_tag_index()
                else:
                    pass

            case _:
                raise ValueError(
                    f"Invalid operation: {operation}. Must be one of {valid_operations}"
                )

    except Exception as exc:
        raise Exception(
            f"Unhandled exception configuring tag_index. Details: {exc}"
        )

default_timeout()

Return the default timeout period.

Returns:

Type	Description
int	The number of seconds in 24 hours

Source code in src/red_utils/ext/diskcache_utils/classes.py

def default_timeout() -> int:
    """Return the default timeout period.

    Returns:
        (int): The number of seconds in 24 hours

    """
    timeout = convert_to_seconds(amount=24, unit="hours")
    return timeout