# Copyright (c) 2023 - 2025, AG2ai, Inc., AG2ai open-source projects maintainers and core contributors # # SPDX-License-Identifier: Apache-2.0 # # Portions derived from https://github.com/microsoft/autogen are under the MIT License. # SPDX-License-Identifier: MIT import sys from types import TracebackType from typing import Any, Optional, Union import diskcache from .abstract_cache_base import AbstractCache if sys.version_info >= (3, 11): from typing import Self else: from typing_extensions import Self class DiskCache(AbstractCache): """Implementation of AbstractCache using the DiskCache library. This class provides a concrete implementation of the AbstractCache interface using the diskcache library for caching data on disk. Attributes: cache (diskcache.Cache): The DiskCache instance used for caching. Methods: __init__(self, seed): Initializes the DiskCache with the given seed. get(self, key, default=None): Retrieves an item from the cache. set(self, key, value): Sets an item in the cache. close(self): Closes the cache. __enter__(self): Context management entry. __exit__(self, exc_type, exc_value, traceback): Context management exit. """ def __init__(self, seed: Union[str, int]): """Initialize the DiskCache instance. Args: seed (Union[str, int]): A seed or namespace for the cache. This is used to create a unique storage location for the cache data. """ self.cache = diskcache.Cache(seed) def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]: """Retrieve an item from the cache. Args: key (str): The key identifying the item in the cache. default (optional): The default value to return if the key is not found. Defaults to None. Returns: The value associated with the key if found, else the default value. """ return self.cache.get(key, default) def set(self, key: str, value: Any) -> None: """Set an item in the cache. Args: key (str): The key under which the item is to be stored. value: The value to be stored in the cache. """ self.cache.set(key, value) def close(self) -> None: """Close the cache. Perform any necessary cleanup, such as closing file handles or releasing resources. """ self.cache.close() def __enter__(self) -> Self: """Enter the runtime context related to the object. Returns: self: The instance itself. """ return self def __exit__( self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback: Optional[TracebackType], ) -> None: """Exit the runtime context related to the object. Perform cleanup actions such as closing the cache. Args: exc_type: The exception type if an exception was raised in the context. exc_value: The exception value if an exception was raised in the context. traceback: The traceback if an exception was raised in the context. """ self.close()