Spaces:
Running
Running
File size: 1,653 Bytes
e70c280 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | """
File: /cache.py
Created Date: Sunday October 19th 2025
Author: Christian Nonis <alch.infoemail@gmail.com>
-----
Last Modified: Sunday October 19th 2025 9:00:36 am
Modified By: the developer formerly known as Christian Nonis at <alch.infoemail@gmail.com>
-----
"""
from typing import Optional
from .interfaces.cache import CacheClient
class CacheAdapter:
"""
Adapter for the cache client.
"""
def __init__(self):
self.cache = None
def add_client(self, client: CacheClient) -> None:
"""
Add a cache client to the adapter.
"""
self.cache = client
def get(self, key: str, brain_id: str = "default") -> str:
"""
Get a value from the cache.
"""
return self.cache.get(key, brain_id)
def set(
self,
key: str,
value: str,
brain_id: str = "default",
expires_in: Optional[int] = None,
) -> bool:
"""
Set a value in the cache with an expiration time.
"""
return self.cache.set(key, value, brain_id, expires_in)
def delete(self, key: str, brain_id: str = "default") -> bool:
"""
Delete a value from the cache.
"""
return self.cache.delete(key, brain_id)
def get_task_keys(self, brain_id: str = "default") -> list[str]:
"""
Get all task keys for a given brain_id.
"""
return self.cache.get_task_keys(brain_id)
def get_task(self, task_id: str, brain_id: str = "default") -> Optional[str]:
"""
Get a specific task by ID.
"""
return self.cache.get_task(task_id, brain_id)
|