Skip to main content

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.