momento_cachedb

langroid/cachedb/momento_cachedb.py

MomentoCacheConfig

Bases: BaseModel

Configuration model for Momento Cache.

MomentoCache(config)

Bases: CacheDB

Momento implementation of the CacheDB.

Parameters:

Name	Type	Description	Default
config	MomentoCacheConfig	The configuration to use.	required

Source code in langroid/cachedb/momento_cachedb.py

def __init__(self, config: MomentoCacheConfig):
    """
    Initialize a MomentoCache with the given config.

    Args:
        config (MomentoCacheConfig): The configuration to use.
    """
    self.config = config
    load_dotenv()

    momento_token = os.getenv("MOMENTO_AUTH_TOKEN")
    if momento_token is None:
        raise ValueError("""MOMENTO_AUTH_TOKEN not set in .env file""")
    else:
        self.client = momento.CacheClient(
            configuration=momento.Configurations.Laptop.v1(),
            credential_provider=momento.CredentialProvider.from_environment_variable(
                "MOMENTO_AUTH_TOKEN"
            ),
            default_ttl=timedelta(seconds=self.config.ttl),
        )
        self.client.create_cache(self.config.cachename)

clear()

Clear keys from current db.

Source code in langroid/cachedb/momento_cachedb.py

def clear(self) -> None:
    """Clear keys from current db."""
    self.client.flush_cache(self.config.cachename)

store(key, value)

Store a value associated with a key.

Parameters:

Name	Type	Description	Default
key	str	The key under which to store the value.	required
value	Any	The value to store.	required

Source code in langroid/cachedb/momento_cachedb.py

def store(self, key: str, value: Any) -> None:
    """
    Store a value associated with a key.

    Args:
        key (str): The key under which to store the value.
        value (Any): The value to store.
    """
    self.client.set(self.config.cachename, key, json.dumps(value))

retrieve(key)

Retrieve the value associated with a key.

Parameters:

Name	Type	Description	Default
key	str	The key to retrieve the value for.	required

Returns:

Name	Type	Description
dict	Dict[str, Any] | str | None	The value associated with the key.

Source code in langroid/cachedb/momento_cachedb.py

def retrieve(self, key: str) -> Dict[str, Any] | str | None:
    """
    Retrieve the value associated with a key.

    Args:
        key (str): The key to retrieve the value for.

    Returns:
        dict: The value associated with the key.
    """
    value = self.client.get(self.config.cachename, key)
    if isinstance(value, CacheGet.Hit):
        return json.loads(value.value_string)  # type: ignore
    else:
        return None

delete_keys(keys)

Delete the keys from the cache.

Parameters:

Name	Type	Description	Default
keys	List[str]	The keys to delete.	required

Source code in langroid/cachedb/momento_cachedb.py

def delete_keys(self, keys: List[str]) -> None:
    """
    Delete the keys from the cache.

    Args:
        keys (List[str]): The keys to delete.
    """
    for key in keys:
        self.client.delete(self.config.cachename, key)

delete_keys_pattern(pattern)

Delete the keys from the cache with the given pattern.

Parameters:

Name	Type	Description	Default
prefix	str	The pattern to match.	required

Source code in langroid/cachedb/momento_cachedb.py

def delete_keys_pattern(self, pattern: str) -> None:
    """
    Delete the keys from the cache with the given pattern.

    Args:
        prefix (str): The pattern to match.
    """
    raise NotImplementedError(
        """
        MomentoCache does not support delete_keys_pattern.
        Please use RedisCache instead.
        """
    )