aixplain.utils.cache_utils

get_cache_expiry

def get_cache_expiry() -> int

[view_source]

Get the cache expiration duration in seconds.

Retrieves the cache expiration duration from the CACHE_EXPIRY_TIME environment variable. If not set, falls back to the default CACHE_DURATION.

Returns:

• int - The cache expiration duration in seconds.

save_to_cache

def save_to_cache(cache_file: str, data: dict, lock_file: str) -> None

[view_source]

Save data to a cache file with thread-safe file locking.

This function saves the provided data to a JSON cache file along with a timestamp. It uses file locking to ensure thread safety during writing.

Arguments:

• cache_file str - Path to the cache file where data will be saved.
• data dict - The data to be cached. Must be JSON-serializable.
• lock_file str - Path to the lock file used for thread safety.

Notes:

• Creates the cache directory if it doesn't exist
• Logs an error if saving fails but doesn't raise an exception
• The data is saved with a timestamp for expiration checking

load_from_cache

def load_from_cache(cache_file: str, lock_file: str) -> dict

[view_source]

Load data from a cache file with expiration checking.

This function loads data from a JSON cache file if it exists and hasn't expired. It uses file locking to ensure thread safety during reading.

Arguments:

• cache_file str - Path to the cache file to load data from.
• lock_file str - Path to the lock file used for thread safety.

Returns:

• dict - The cached data if the cache exists and hasn't expired, None otherwise.

Notes:

• Returns None if the cache file doesn't exist
• Returns None if the cached data has expired based on CACHE_EXPIRY_TIME
• Uses thread-safe file locking for reading