cache.cosmos_db_cache

​

CosmosDBCache

class CosmosDBCache(AbstractCache)

Synchronous implementation of AbstractCache using Azure Cosmos DB NoSQL API.

This class provides a concrete implementation of the AbstractCache interface using Azure Cosmos DB for caching data, with synchronous operations.

Attributes:

• seed Union[str, int] - A seed or namespace used as a partition key.
• client CosmosClient - The Cosmos DB client used for caching.
• container - The container instance used for caching.

​

__init__

def __init__(seed: Union[str, int], cosmosdb_config: CosmosDBConfig)

Initialize the CosmosDBCache instance.

Arguments:

• seed Union[str, int] - A seed or namespace for the cache, used as a partition key.
• connection_string str - The connection string for the Cosmos DB account.
• container_id str - The container ID to be used for caching.
• client Optional[CosmosClient] - An existing CosmosClient instance to be used for caching.

​

create_cache

@classmethod
def create_cache(cls, seed: Union[str, int], cosmosdb_config: CosmosDBConfig)

Factory method to create a CosmosDBCache instance based on the provided configuration. This method decides whether to use an existing CosmosClient or create a new one.

​

get

def get(key: str, default: Optional[Any] = None) -> Optional[Any]

Retrieve an item from the Cosmos DB cache.

Arguments:

• key str - The key identifying the item in the cache.
• default optional - The default value to return if the key is not found.

Returns:

The deserialized value associated with the key if found, else the default value.

​

set

def set(key: str, value: Any) -> None

Set an item in the Cosmos DB cache.

Arguments:

• key str - The key under which the item is to be stored.
• value - The value to be stored in the cache.

Notes:

The value is serialized using pickle before being stored.

​

close

def close() -> None

Close the Cosmos DB client.

Perform any necessary cleanup, such as closing network connections.

​

__enter__

def __enter__()

Context management entry.

Returns:

• self - The instance itself.

​

__exit__

def __exit__(exc_type: Optional[type], exc_value: Optional[Exception],
             traceback: Optional[Any]) -> None

Context management exit.

Perform cleanup actions such as closing the Cosmos DB client.