Skip to content

Cache

autogen.Cache #

Cache(config)

Bases: AbstractCache

A wrapper class for managing cache configuration and instances.

This class provides a unified interface for creating and interacting with different types of cache (e.g., Redis, Disk). It abstracts the underlying cache implementation details, providing methods for cache operations.

ATTRIBUTE DESCRIPTION
config

A dictionary containing cache configuration.

TYPE: Dict[str, Any]

cache

The cache instance created based on the provided configuration.

Initialize the Cache with the given configuration.

Validates the configuration keys and creates the cache instance.

PARAMETER DESCRIPTION
config

A dictionary containing the cache configuration.

TYPE: Dict[str, Any]

RAISES DESCRIPTION
ValueError

If an invalid configuration key is provided.
Source code in autogen/cache/cache.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
def __init__(self, config: dict[str, Any]):
    """Initialize the Cache with the given configuration.

    Validates the configuration keys and creates the cache instance.

    Args:
        config (Dict[str, Any]): A dictionary containing the cache configuration.

    Raises:
        ValueError: If an invalid configuration key is provided.
    """
    self.config = config
    # Ensure that the seed is always treated as a string before being passed to any cache factory or stored.
    self.config["cache_seed"] = str(self.config.get("cache_seed", 42))

    # validate config
    for key in self.config:
        if key not in self.ALLOWED_CONFIG_KEYS:
            raise ValueError(f"Invalid config key: {key}")
    # create cache instance
    self.cache = CacheFactory.cache_factory(
        seed=self.config["cache_seed"],
        redis_url=self.config.get("redis_url"),
        cache_path_root=self.config.get("cache_path_root"),
        cosmosdb_config=self.config.get("cosmos_db_config"),
    )

ALLOWED_CONFIG_KEYS class-attribute instance-attribute #

ALLOWED_CONFIG_KEYS = ['cache_seed', 'redis_url', 'cache_path_root', 'cosmos_db_config']

config instance-attribute #

config = config

cache instance-attribute #

cache = cache_factory(seed=config['cache_seed'], redis_url=get('redis_url'), cache_path_root=get('cache_path_root'), cosmosdb_config=get('cosmos_db_config'))

redis staticmethod #

redis(cache_seed=42, redis_url='redis://localhost:6379/0')

Create a Redis cache instance.

PARAMETER DESCRIPTION
cache_seed

A seed for the cache. Defaults to 42.

TYPE: Union[str, int] DEFAULT: 42

redis_url

The URL for the Redis server. Defaults to "redis://localhost:6379/0".

TYPE: str DEFAULT: 'redis://localhost:6379/0'

RETURNS DESCRIPTION
Cache

A Cache instance configured for Redis.

TYPE: Cache

Source code in autogen/cache/cache.py 
37
38
39
40
41
42
43
44
45
46
47
48
@staticmethod
def redis(cache_seed: Union[str, int] = 42, redis_url: str = "redis://localhost:6379/0") -> Cache:
    """Create a Redis cache instance.

    Args:
        cache_seed (Union[str, int], optional): A seed for the cache. Defaults to 42.
        redis_url (str, optional): The URL for the Redis server. Defaults to "redis://localhost:6379/0".

    Returns:
        Cache: A Cache instance configured for Redis.
    """
    return Cache({"cache_seed": cache_seed, "redis_url": redis_url})

disk staticmethod #

disk(cache_seed=42, cache_path_root='.cache')

Create a Disk cache instance.

PARAMETER DESCRIPTION
cache_seed

A seed for the cache. Defaults to 42.

TYPE: Union[str, int] DEFAULT: 42

cache_path_root

The root path for the disk cache. Defaults to ".cache".

TYPE: str DEFAULT: '.cache'

RETURNS DESCRIPTION
Cache

A Cache instance configured for Disk caching.

TYPE: Cache

Source code in autogen/cache/cache.py 
50
51
52
53
54
55
56
57
58
59
60
61
@staticmethod
def disk(cache_seed: Union[str, int] = 42, cache_path_root: str = ".cache") -> Cache:
    """Create a Disk cache instance.

    Args:
        cache_seed (Union[str, int], optional): A seed for the cache. Defaults to 42.
        cache_path_root (str, optional): The root path for the disk cache. Defaults to ".cache".

    Returns:
        Cache: A Cache instance configured for Disk caching.
    """
    return Cache({"cache_seed": cache_seed, "cache_path_root": cache_path_root})

cosmos_db staticmethod #

cosmos_db(connection_string=None, container_id=None, cache_seed=42, client=None)

Create a Cosmos DB cache instance with 'autogen_cache' as database ID.

PARAMETER DESCRIPTION
connection_string

Connection string to the Cosmos DB account.

TYPE: str DEFAULT: None

container_id

The container ID for the Cosmos DB account.

TYPE: str DEFAULT: None

cache_seed

A seed for the cache.

TYPE: Union[str, int] DEFAULT: 42

client

Optional[CosmosClient]: Pass an existing Cosmos DB client.

TYPE: Optional[Any] DEFAULT: None

RETURNS DESCRIPTION
Cache

A Cache instance configured for Cosmos DB.

TYPE: Cache

Source code in autogen/cache/cache.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
@staticmethod
def cosmos_db(
    connection_string: Optional[str] = None,
    container_id: Optional[str] = None,
    cache_seed: Union[str, int] = 42,
    client: Optional[Any] = None,
) -> Cache:
    """Create a Cosmos DB cache instance with 'autogen_cache' as database ID.

    Args:
        connection_string (str, optional): Connection string to the Cosmos DB account.
        container_id (str, optional): The container ID for the Cosmos DB account.
        cache_seed (Union[str, int], optional): A seed for the cache.
        client: Optional[CosmosClient]: Pass an existing Cosmos DB client.

    Returns:
        Cache: A Cache instance configured for Cosmos DB.
    """
    cosmos_db_config = {
        "connection_string": connection_string,
        "database_id": "autogen_cache",
        "container_id": container_id,
        "client": client,
    }
    return Cache({"cache_seed": str(cache_seed), "cosmos_db_config": cosmos_db_config})

get #

get(key, default=None)

Retrieve an item from the cache.

PARAMETER DESCRIPTION
key

The key identifying the item in the cache.

TYPE: str

default

The default value to return if the key is not found. Defaults to None.

TYPE: optional DEFAULT: None

RETURNS DESCRIPTION
Optional[Any]

The value associated with the key if found, else the default value.
Source code in autogen/cache/cache.py 
142
143
144
145
146
147
148
149
150
151
152
153
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)

set #

set(key, value)

Set an item in the cache.

PARAMETER DESCRIPTION
key

The key under which the item is to be stored.

TYPE: str

value

The value to be stored in the cache.

TYPE: Any

Source code in autogen/cache/cache.py 
155
156
157
158
159
160
161
162
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)

close #

close()

Close the cache.

Perform any necessary cleanup, such as closing connections or releasing resources.

Source code in autogen/cache/cache.py 
164
165
166
167
168
169
def close(self) -> None:
    """Close the cache.

    Perform any necessary cleanup, such as closing connections or releasing resources.
    """
    self.cache.close()