Cachify Context

The cachify context provides a simple interface for managing caching operations. Generally speaking, unless you need to customize your cachify context, a single instance of the cachify context is sufficient for most use cases.

---

API Reference

Bases: abc.ABC

The CacheContext is used to manage the cache context

Initializes the CacheContext

Source code in kvdb/io/cachify/cache.py

def __init__(
    self,
    cache_name: Optional[str] = None,
    cachify_class: Optional[Type['Cachify']] = None,
    session: Optional['KVDBSession'] = None,
    session_name: Optional[str] = None,
    partial_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs,
):
    """
    Initializes the CacheContext
    """
    from kvdb.configs import settings

    self.settings = settings.model_copy()
    self.config = self.settings.cache
    self.cache_name = cache_name
    self._session: Optional['KVDBSession'] = session
    self._session_name: Optional[str] = session_name

    self.cachify_contexts: Dict[str, 'Cachify'] = {}
    self.partial_kwargs = partial_kwargs or {}

    self._ctx_available: Optional[bool] = None

    cache_config, kwargs = self.config.extract_config_and_kwargs(**kwargs)
    self.config.update_config(**cache_config)
    self.configure_classes(cachify_class = cachify_class, is_init = True)
    self.registered_cachify_object: Dict[str, Dict[str, Dict]] = {}
    self.registered_cachify_validation_func: Dict[str, str] = {}
    self.logger = self.settings.logger
    self.autologger = self.settings.autologger
    self.verbose: Optional[bool] = kwargs.get('verbose', self.settings.debug_enabled)
    # self.has_async_loop = self.settings.is_in_async_loop()
    self._kwargs = kwargs

ctx property

ctx: typing.Optional['KVDBSession']

Returns the KV Session if it is available

has_async_loop property

has_async_loop: bool

Checks if the current process is running in an async loop

session property

session: 'KVDBSession'

Returns the session

add_function

add_function(
    function: typing.Union[typing.Callable, str],
    function_name: typing.Optional[str] = None,
    **kwargs
) -> kvdb.io.cachify.base.Cachify

Adds a function to the cachify context

Source code in kvdb/io/cachify/cache.py

def add_function(
    self,
    function: Union[Callable, str],
    function_name: Optional[str] = None,
    **kwargs
) -> Cachify:
    """
    Adds a function to the cachify context
    """
    cachify = self.create_cachify(
        function_name = function_name or get_function_name(function),
        **kwargs
    )
    if cachify.function_name not in self.cachify_contexts:
        self.cachify_contexts[cachify.function_name] = cachify
    return self.cachify_contexts[cachify.function_name]

bind_session

bind_session(session: 'KVDBSession')

Binds the session

Source code in kvdb/io/cachify/cache.py

def bind_session(self, session: 'KVDBSession'):
    """
    Binds the session
    """
    self._session = session

configure_classes

configure_classes(
    cachify_class: typing.Optional[
        typing.Type["Cachify"]
    ] = None,
    is_init: typing.Optional[bool] = False,
)

Configures the classes

Source code in kvdb/io/cachify/cache.py

def configure_classes(
    self,
    cachify_class: Optional[Type['Cachify']] = None,
    is_init: Optional[bool] = False,
):
    """
    Configures the classes
    """
    if cachify_class is None and is_init:
        cachify_class = Cachify
    elif cachify_class and isinstance(cachify_class, str):
        cachify_class = lazy_import(cachify_class)
    if cachify_class is not None:
        self.cachify_class = cachify_class

create_cachify

create_cachify(**kwargs) -> 'Cachify'

Creates a cachify object

Source code in kvdb/io/cachify/cache.py

def create_cachify(self, **kwargs) -> 'Cachify':
    """
    Creates a cachify object
    """
    base_kwargs = self.config.model_dump(exclude_none=True)
    base_kwargs.update(kwargs)
    base_kwargs.update(self.partial_kwargs)
    # self.logger.debug(f'Creating Cachify with kwargs: {base_kwargs}')
    return self.cachify_class(session = self.session, settings = self.settings, **kwargs)

register

register(
    function: typing.Optional[
        kvdb.io.cachify.base.FunctionT
    ] = None,
    **kwargs
) -> typing.Callable[
    [kvdb.io.cachify.base.FunctionT],
    kvdb.io.cachify.base.FunctionT,
]

Registers a function to cachify

Source code in kvdb/io/cachify/cache.py

def register(
    self,
    function: Optional[FunctionT] = None,
    **kwargs,
) -> Callable[[FunctionT], FunctionT]:
    """
    Registers a function to cachify
    """
    if function is not None:
        if is_uninit_method(function):
            return self.register_object_method(**kwargs)(function)
        cachify = self.add_function(
            function = function,
            **kwargs,
        )
        return cachify(function)

    def decorator(func: FunctionT) -> Callable[..., ReturnValueT]:
        """
        The decorator
        """
        if is_uninit_method(func):
            return self.register_object_method(**kwargs)(func)
        cachify = self.add_function(
            function = func,
            **kwargs,
        )
        return cachify(func)
    return decorator

register_object

register_object(
    validator_function_name: typing.Optional[str] = None,
    debug_enabled: typing.Optional[bool] = None,
    **_kwargs
) -> types.ModuleType

Register the underlying object

Source code in kvdb/io/cachify/cache.py

def register_object(
    self, 
    validator_function_name: Optional[str] = None,
    debug_enabled: Optional[bool] = None,
    **_kwargs
) -> ModuleType:
    """
    Register the underlying object
    """
    partial_kws = {k:v for k,v in _kwargs.items() if v is not None}
    validator_function_name = validator_function_name or 'validate_cachify'
    autologger = logger if debug_enabled else null_logger

    def object_decorator(obj: ModuleType) -> ModuleType:
        """
        The decorator that patches the object
        """
        _obj_id = f'{obj.__module__}.{obj.__name__}'
        autologger.info(f'Registering |g|{_obj_id}|e|', colored = True)
        patch_object_for_kvdb(obj)

        # create_register_abstract_object_subclass_function(obj)
        if _obj_id not in self.registered_cachify_object: self.registered_cachify_object[_obj_id] = {}
        if _obj_id not in self.registered_cachify_validation_func: self.registered_cachify_validation_func[_obj_id] = validator_function_name

        if not hasattr(obj, '__cachify_init__'):

            def __cachify_init__(_self, obj_id: str, *args, **kwargs):
                """
                Initializes the object
                """
                parent_obj_names = get_parent_object_class_names(obj, _obj_id)
                __obj_id = f'{_self.__class__.__module__}.{_self.__class__.__name__}'
                __obj_bases = [f'{base.__module__}.{base.__name__}' for base in _self.__class__.__bases__]
                __obj_bases = [o for o in __obj_bases if o != __obj_id and o not in parent_obj_names]

                autologger.info(f'Initializing |g|{__obj_id}|e| with bases: {__obj_bases}', colored = True)
                autologger.info(f'Parent Objects: {parent_obj_names}, for {_obj_id}, {_self.__cachify_subcls__}', prefix = __obj_id, colored = True)

                cachify_functions = {}

                # Register the parent functions first
                parent_obj_functions = self.registered_cachify_object[obj_id]
                parent_obj_validator_func = getattr(_self, self.registered_cachify_validation_func[obj_id], None)
                cachify_validators = {
                    f: parent_obj_validator_func
                    for f in parent_obj_functions
                }

                cachify_functions.update(parent_obj_functions)

                # Handle Subclass level objects
                for sub_obj_id in _self.__cachify_subcls__:
                    if sub_obj_id not in self.registered_cachify_object: continue
                    subcls_functions = self.registered_cachify_object[sub_obj_id]
                    subcls_validator_func = getattr(_self, self.registered_cachify_validation_func[sub_obj_id], None)
                    cachify_validators.update({
                        f: subcls_validator_func
                        for f in subcls_functions
                    })
                    cachify_functions.update(subcls_functions)

                # Handle Base Class level objects
                for base_obj_id in __obj_bases:
                    if base_obj_id not in self.registered_cachify_object: continue
                    base_obj_functions = self.registered_cachify_object[base_obj_id]
                    base_obj_validator_func = getattr(_self, self.registered_cachify_validation_func[base_obj_id], None)
                    cachify_validators.update({
                        f: base_obj_validator_func
                        for f in base_obj_functions
                    })
                    cachify_functions.update(base_obj_functions)

                # Now we do the actual patching
                for func, task_partial_kws in cachify_functions.items():
                    if not hasattr(_self, func): 
                        autologger.info(f'Skipping {func} for {__obj_id}')
                        continue

                    autologger.info(f'Patching {func} for {__obj_id}')
                    func_kws = partial_kws.copy()
                    func_kws.update(task_partial_kws)

                    if cachify_validators[func] is not None:
                        func_kws = cachify_validators[func](func, **func_kws)
                        if func_kws is None: continue

                    if 'function_name' not in func_kws:
                        func_kws['function_name'] = f'{_self.__class__.__name__}.{func}'

                    patched_func = self.register(function = getattr(_self, func), **func_kws)
                    setattr(_self, func, patched_func)


            setattr(obj, '__cachify_init__', __cachify_init__)
            setattr(obj, '__cachify_subcls__', [])
            obj.__kvdb_initializers__.append('__cachify_init__')
        else:
            obj.__cachify_subcls__.append(_obj_id)

        return obj
    return object_decorator

register_object_method

register_object_method(
    **kwargs,
) -> typing.Callable[
    [kvdb.io.cachify.base.FunctionT],
    kvdb.io.cachify.base.FunctionT,
]

Registers an object method function to be cached

Source code in kvdb/io/cachify/cache.py

def register_object_method(self, **kwargs) -> Callable[[FunctionT], FunctionT]:
    """
    Registers an object method function to be cached
    """
    kwargs = {k:v for k,v in kwargs.items() if v is not None}

    def decorator(func: FunctionT) -> Callable[..., ReturnValueT]:
        """
        The decorator
        """
        task_obj_id = f'{func.__module__}.{func.__qualname__.split(".")[0]}'
        if task_obj_id not in self.registered_cachify_object:
            self.registered_cachify_object[task_obj_id] = {}
        func_name = func.__name__
        if func_name not in self.registered_cachify_object[task_obj_id]:
            self.registered_cachify_object[task_obj_id][func_name] = kwargs
        return func
    return decorator

safely

safely(timeout: typing.Optional[float] = 2.0)

Safely wraps the function

Source code in kvdb/io/cachify/cache.py

@contextlib.contextmanager
def safely(self, timeout: Optional[float] = 2.0):
    """
    Safely wraps the function
    """
    if self.has_async_loop:
        with anyio.move_on_after(timeout = timeout):
            yield
    else:
        with timeout_ctx(timeout, raise_errors = False):
            yield