Skip to content

Context Manager

The cachify context manager registers and manages all caching operations.

The following example demonstrates usage of the Cachify class as a context manager.

context.py
"""
A Very basic demonstration of how to use cachify with objects using a context

Similiar to registering tasks, a context can be created to register objects or functions, where
the passed parameters will be used as the default parameters for all registered functions.

Run this example:

    # cwd: examples/caching
    $ python context.py
"""

import time
import abc
import asyncio
from lazyops.utils.times import Timer
from kvdb.io import cachify
from kvdb.utils.logs import logger

DEBUG_ENABLED = False

cache_context = cachify.create_context(
    cache_name='test', # Unique name for the cache. This is used to identify the cache in the cache registry
    prefix = '_kvtest_', # Prefix for the cache key
    ttl = 10,
    verbosity = 2 if DEBUG_ENABLED else None,
    cache_max_size = 15,
)

@cache_context.register_object()
class TestObject(abc.ABC):

    def __init__(self, *args, **kwargs):
        logger.info('running init')

    @cache_context.register()
    async def async_fibonacci(self, number: int):
        if number == 0: return 0
        elif number == 1: return 1
        return await self.async_fibonacci(number - 1) + await self.async_fibonacci(number - 2)

    @cache_context.register()
    def fibonacci(self, number: int):
        if number == 0: return 0
        elif number == 1: return 1
        return self.fibonacci(number - 1) + self.fibonacci(number - 2)

    # No Cache Versions
    async def async_fibonacci_nc(self, number: int):
        if number == 0: return 0
        elif number == 1: return 1
        return await self.async_fibonacci_nc(number - 1) + await self.async_fibonacci_nc(number - 2)

    def fibonacci_nc(self, number: int):
        if number == 0: return 0
        elif number == 1: return 1
        return self.fibonacci_nc(number - 1) + self.fibonacci_nc(number - 2)


async def run_tests(
    start_n: int = 1,
    runs: int = 10,
    print_every: int = 5,
):

    """
    Test that both results are the same.
    """

    t = Timer(format_ms=True)
    o = TestObject()

    # Test Sync
    st = Timer(format_ms=True)
    for i in range(runs):
        r = o.fibonacci(start_n+i)
        d = st.duration_s
        if i % print_every == 0:
            logger.info(f'[Sync - {i}/{runs}] Result: {r} | Time: {d}')
    logger.info(f'[Sync] Cache Average Time: {st.total_average_s(runs)} | Total Time: {st.total_s}')
    logger.info(o.fibonacci.cache_info(), prefix = '[Sync] Cache Info')

    # Test Async
    at = Timer(format_ms=True)
    for i in range(runs):
        r = await o.async_fibonacci(start_n+i)
        d = at.duration_s
        if i % print_every == 0:
            logger.info(f'[Async - {i}/{runs}] Result: {r} | Time: {d}')
    logger.info(f'[Async] Cache Average Time: {at.total_average_s(runs)} | Total Time: {at.total_s}')
    logger.info(await o.async_fibonacci.cache_info(), prefix = '[Async] Cache Info')
    logger.info(t.total_s, prefix = 'Total Time')

    # Clear the Cache
    o.fibonacci.clear()
    logger.info(o.fibonacci.cache_info(), prefix = '[Sync] Cache Info')

    await o.async_fibonacci.clear()
    logger.info(await o.async_fibonacci.cache_info(), prefix = '[Async] Cache Info')

    logger.info('Testing Non-Cached Functions')
    t = Timer(format_ms=True)

    # Test Sync
    st = Timer(format_ms=True)
    for i in range(runs):
        r = o.fibonacci_nc(start_n+i)
        d = st.duration_s
        if i % print_every == 0:
            logger.info(f'[Sync - {i}/{runs}] Result: {r} | Time: {d}')
    logger.info(f'[Sync] Cache Average Time: {st.total_average_s(runs)} | Total Time: {st.total_s}')

    # Test Async
    at = Timer(format_ms=True)
    for i in range(runs):
        r = await o.async_fibonacci_nc(start_n+i)
        d = at.duration_s
        if i % print_every == 0:
            logger.info(f'[Async - {i}/{runs}] Result: {r} | Time: {d}')
    logger.info(f'[Async] Cache Average Time: {at.total_average_s(runs)} | Total Time: {at.total_s}')
    logger.info(t.total_s, prefix = 'Total Time')



if __name__ == '__main__':
    asyncio.run(run_tests(
        start_n = 5,
        runs = 40,
        print_every = 5,
    ))

API Reference

Bases: abc.ABC

The CachifyManager handles management of the Cachify component

Initializes the CachifyManager

METHOD DESCRIPTION
configure_classes

Configures the classes
create_cachify_context

Creates a CachifyContext
get_cachify_context

Gets a CachifyContext
register

Registers a function with the CachifyManager
register_object

Register the underlying object
register_object_method

Register the underlying object
Source code in kvdb/io/cachify/main.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
def __init__(self, *args, **kwargs):
    """
    Initializes the CachifyManager
    """
    from kvdb.configs import settings
    self.settings: 'KVDBSettings' = settings

    self.cachify_context_class: Type[CachifyContext] = CachifyContext
    self.cachify_contexts: Dict[str, CachifyContext] = {}

    self.cachify_class: Type[Cachify] = Cachify
    self.logger = self.settings.logger
    self.autologger = self.settings.autologger

configure_classes 

configure_classes(
    cachify_context_class: typing.Optional[
        typing.Type[kvdb.io.cachify.cache.CachifyContext]
    ] = None,
    cachify_class: typing.Optional[
        typing.Type[kvdb.io.cachify.base.Cachify]
    ] = None,
) -> None

Configures the classes

Source code in kvdb/io/cachify/main.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def configure_classes(
    self,
    cachify_context_class: Optional[Type[CachifyContext]] = None,
    cachify_class: Optional[Type[Cachify]] = None,
) -> None:
    """
    Configures the classes
    """
    if cachify_context_class and isinstance(cachify_context_class, str):
        cachify_context_class = lazy_import(cachify_context_class)
    if cachify_context_class is not None:
        self.cachify_context_class = cachify_context_class
    if 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_context 

create_cachify_context(
    cache_name: typing.Optional[str] = None,
    cachify_class: typing.Optional[
        typing.Type["Cachify"]
    ] = None,
    cachify_context_class: typing.Optional[
        typing.Type[kvdb.io.cachify.cache.CachifyContext]
    ] = None,
    session: typing.Optional["KVDBSession"] = None,
    session_name: typing.Optional[str] = None,
    partial_kwargs: typing.Optional[
        typing.Dict[str, typing.Any]
    ] = None,
    **kwargs
) -> kvdb.io.cachify.cache.CachifyContext

Creates a CachifyContext

Source code in kvdb/io/cachify/main.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
def create_cachify_context(
    self,
    cache_name: Optional[str] = None,
    cachify_class: Optional[Type['Cachify']] = None,
    cachify_context_class: Optional[Type[CachifyContext]] = None,
    session: Optional['KVDBSession'] = None,
    session_name: Optional[str] = None,
    partial_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs,
) -> CachifyContext:
    """
    Creates a CachifyContext
    """
    cache_name = cache_name or self.default_cache_name
    if cache_name not in self.cachify_contexts:
        cachify_context_class = cachify_context_class or self.cachify_context_class
        self.cachify_contexts[cache_name] = cachify_context_class(
            cache_name = cache_name,
            cachify_class = cachify_class or self.cachify_class,
            session = session,
            session_name = session_name,
            partial_kwargs = partial_kwargs,
            **kwargs,
        )
    return self.cachify_contexts[cache_name]

get_cachify_context 

get_cachify_context(
    cache_name: typing.Optional[str] = None, **kwargs
) -> kvdb.io.cachify.cache.CachifyContext

Gets a CachifyContext

Source code in kvdb/io/cachify/main.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
def get_cachify_context(
    self,
    cache_name: Optional[str] = None,
    **kwargs,
) -> CachifyContext:
    """
    Gets a CachifyContext
    """
    cache_name = cache_name or self.default_cache_name
    if cache_name not in self.cachify_contexts:
        self.create_cachify_context(cache_name = cache_name, **kwargs)
    return self.cachify_contexts[cache_name]

register 

register(
    function: typing.Optional[
        kvdb.io.cachify.base.FunctionT
    ] = None,
    cache_name: typing.Optional[str] = None,
    ttl: typing.Optional[int] = 60 * 10,
    ttl_kws: typing.Optional[typing.List[str]] = [
        "cache_ttl"
    ],
    keybuilder: typing.Optional[typing.Callable] = None,
    name: typing.Optional[
        typing.Union[str, typing.Callable]
    ] = None,
    typed: typing.Optional[bool] = True,
    exclude_keys: typing.Optional[typing.List[str]] = None,
    exclude_null: typing.Optional[bool] = True,
    exclude_exceptions: typing.Optional[
        typing.Union[bool, typing.List[Exception]]
    ] = True,
    prefix: typing.Optional[str] = "_kvc_",
    exclude_null_values_in_hash: typing.Optional[
        bool
    ] = None,
    exclude_default_values_in_hash: typing.Optional[
        bool
    ] = None,
    disabled: typing.Optional[
        typing.Union[bool, typing.Callable]
    ] = None,
    disabled_kws: typing.Optional[typing.List[str]] = [
        "cache_disable"
    ],
    invalidate_after: typing.Optional[
        typing.Union[int, typing.Callable]
    ] = None,
    invalidate_if: typing.Optional[typing.Callable] = None,
    invalidate_kws: typing.Optional[typing.List[str]] = [
        "cache_invalidate"
    ],
    overwrite_if: typing.Optional[typing.Callable] = None,
    overwrite_kws: typing.Optional[typing.List[str]] = [
        "cache_overwrite"
    ],
    retry_enabled: typing.Optional[bool] = False,
    retry_max_attempts: typing.Optional[int] = 3,
    retry_giveup_callable: typing.Optional[
        typing.Callable[..., bool]
    ] = None,
    timeout: typing.Optional[float] = 5.0,
    verbosity: typing.Optional[int] = None,
    raise_exceptions: typing.Optional[bool] = True,
    encoder: typing.Optional[
        typing.Union[str, typing.Callable]
    ] = None,
    decoder: typing.Optional[
        typing.Union[str, typing.Callable]
    ] = None,
    hit_setter: typing.Optional[typing.Callable] = None,
    hit_getter: typing.Optional[typing.Callable] = None,
    cache_max_size: typing.Optional[int] = None,
    cache_max_size_policy: typing.Optional[
        typing.Union[str, kvdb.types.common.CachePolicy]
    ] = kvdb.types.common.CachePolicy.LFU,
    post_init_hook: typing.Optional[
        typing.Union[str, typing.Callable]
    ] = None,
    post_call_hook: typing.Optional[
        typing.Union[str, typing.Callable]
    ] = None,
    hset_enabled: typing.Optional[bool] = True,
    silenced_stages: typing.Optional[
        typing.List[str]
    ] = None,
    session: typing.Optional["KVDBSession"] = None,
    session_name: typing.Optional[str] = None,
    session_kwargs: typing.Optional[
        typing.Dict[str, typing.Any]
    ] = None,
    **kwargs
) -> typing.Callable[
    [kvdb.io.cachify.base.FunctionT],
    kvdb.io.cachify.base.FunctionT,
]

register(
    cache_name: typing.Optional[str] = None,
    function: typing.Optional[
        kvdb.io.cachify.base.FunctionT
    ] = None,
    session: typing.Optional["KVDBSession"] = None,
    session_name: typing.Optional[str] = None,
    session_kwargs: typing.Optional[
        typing.Dict[str, typing.Any]
    ] = None,
    **kwargs
) -> typing.Callable[
    [kvdb.io.cachify.base.FunctionT],
    kvdb.io.cachify.base.FunctionT,
]

Registers a function with the CachifyManager

PARAMETER DESCRIPTION
cache_name

The name of the cache. Defaults to None.

TYPE: typing.Optional[str] DEFAULT: None

function

The function to register. Defaults to None.

TYPE: typing.Optional[kvdb.io.cachify.base.FunctionT] DEFAULT: None

session

The session to use. Defaults to None.

TYPE: typing.Optional['KVDBSession'] DEFAULT: None

session_name

The session name to use. Defaults to None.

TYPE: typing.Optional[str] DEFAULT: None

session_kwargs

The session kwargs to use. Defaults to None.

TYPE: typing.Optional[typing.Dict[str, typing.Any]] DEFAULT: None

Source code in kvdb/io/cachify/main.py 
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def register(
    self,
    cache_name: Optional[str] = None,
    function: Optional[FunctionT] = None,
    session: Optional['KVDBSession'] = None,
    session_name: Optional[str] = None,
    session_kwargs: Optional[Dict[str, Any]] = None,
    **kwargs,
) -> Callable[[FunctionT], FunctionT]:  # sourcery skip: default-mutable-arg
    """
    Registers a function with the CachifyManager

    Args:
        cache_name (Optional[str], optional): The name of the cache. Defaults to None.
        function (Optional[FunctionT], optional): The function to register. Defaults to None.
        session (Optional['KVDBSession'], optional): The session to use. Defaults to None.
        session_name (Optional[str], optional): The session name to use. Defaults to None.
        session_kwargs (Optional[Dict[str, Any]], optional): The session kwargs to use. Defaults to None.
    """
    cache_name = cache_name or self.default_cache_name
    cachify_context = self.get_cachify_context(
        cache_name = cache_name,
        session = session,
        session_name = session_name,
        session_kwargs = session_kwargs,
        partial_kwargs = kwargs,
    )
    return cachify_context.register(function = function, **kwargs)

register_object 

register_object(
    cache_name: typing.Optional[str] = None, **kwargs
) -> types.ModuleType

Register the underlying object

Source code in kvdb/io/cachify/main.py 
252
253
254
255
256
257
258
259
260
261
262
def register_object(
    self, 
    cache_name: Optional[str] = None, 
    **kwargs
) -> ModuleType:
    """
    Register the underlying object
    """
    cache_name = cache_name or self.default_cache_name
    cachify_context = self.get_cachify_context(cache_name=cache_name, **kwargs)
    return cachify_context.register_object(**kwargs)

register_object_method 

register_object_method(
    cache_name: typing.Optional[str] = None, **kwargs
) -> types.ModuleType

Register the underlying object

Source code in kvdb/io/cachify/main.py 
264
265
266
267
268
269
270
271
272
273
274
def register_object_method(
    self, 
    cache_name: Optional[str] = None, 
    **kwargs
) -> ModuleType:
    """
    Register the underlying object
    """
    cache_name = cache_name or self.default_cache_name
    cachify_context = self.get_cachify_context(cache_name=cache_name, **kwargs)
    return cachify_context.register_object_method(**kwargs)