Spaces:
Running
Running
File size: 1,757 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:01:18 am
Modified By: the developer formerly known as Christian Nonis at <alch.infoemail@gmail.com>
-----
"""
from abc import ABC, abstractmethod
from typing import Optional
class CacheClient(ABC):
"""
Abstract base class for cache clients.
"""
@abstractmethod
def get(self, key: str, brain_id: str) -> str:
"""
Get a value from the cache.
"""
raise NotImplementedError("get method not implemented")
@abstractmethod
def set(
self, key: str, value: str, brain_id: str, expires_in: Optional[int] = None
) -> bool:
"""
Set a value in the cache with an expiration time.
Args:
key: The key to set.
value: The value to set.
expires_in: The expiration time in seconds.
Returns:
True if the value was set, False otherwise.
"""
raise NotImplementedError("set method not implemented")
@abstractmethod
def delete(self, key: str, brain_id: str) -> bool:
"""
Delete a value from the cache.
"""
raise NotImplementedError("delete method not implemented")
@abstractmethod
def get_task_keys(self, brain_id: str) -> list[str]:
"""
Get all task keys for a given brain_id.
"""
raise NotImplementedError("get_task_keys method not implemented")
@abstractmethod
def get_task(self, task_id: str, brain_id: str) -> Optional[str]:
"""
Get a specific task by ID.
"""
raise NotImplementedError("get_task method not implemented")
|