Skip to content

Mixins

Bases: CacheIfMissingMixin

Mixin class which aggregates all mixins from magic_storage.mixins submodule.

Source code in magic_storage/mixins/__init__.py 
 8
 9
10
class FullyFeaturedMixin(CacheIfMissingMixin):
    """Mixin class which aggregates all mixins from magic_storage.mixins
    submodule."""

Bases: ABC

Source code in magic_storage/mixins/_cache_if_missing.py 
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class CacheIfMissingMixin(ABC):
    @abstractmethod
    def is_available(self, __uid: str, /) -> bool:
        ...

    @abstractmethod
    def load_as(  # noqa: FNE004
        self,
        store_type: StoreType,
        uid: str,
        **load_kw: Any,
    ) -> Any:
        ...

    @abstractmethod
    def store_as(
        self,
        store_type: StoreType,
        /,
        uid: str,
        item: Any,
        **dump_kw: Any,
    ) -> str:
        ...

    def cache_if_missing(
        self,
        uid: str,
        callback: Callable[[], _R],
        store_type: StoreType = StoreType.PICKLE,
    ) -> _R:
        """Store and return object if not present in cache, otherwise load from
        cache and return.

        In case of load failure object cache is recreated.

        Parameters
        ----------
        uid : str
            Object identifier used to find object in cache.
        callback : Callable[[], _R]
            Callback function which can create new object if object is not found in cache
        store_type : StoreType, optional
            Determines how object should be stored in cache, by default StoreType.PICKLE

        Returns
        -------
        _R
            Object loaded from cache OR object created with callback and stored to cache.
        """
        uid = make_uid(uid)

        if self.is_available(uid):
            logging.debug(f"'{uid}' is available and will be loaded.")
            try:
                return self.load_as(store_type, uid=uid)  # type: ignore
            except Exception as e:
                logging.exception(e)
            logging.warning(
                f"Failed to load '{uid}' due to loading error. Cache will be recreated."
            )

        else:
            logging.debug(
                f"Resource '{uid}' is NOT available thus will be created."
            )
        # If cache is not present OR if cache load failed
        item = callback()
        self.store_as(store_type, uid=uid, item=item)
        return item  # type: ignore

cache_if_missing(uid, callback, store_type=StoreType.PICKLE) #

Store and return object if not present in cache, otherwise load from cache and return.

In case of load failure object cache is recreated.

Parameters:

Name Type Description Default
uid str

Object identifier used to find object in cache.

 required
callback Callable[[], _R]

Callback function which can create new object if object is not found in cache

 required
store_type StoreType, optional

Determines how object should be stored in cache, by default StoreType.PICKLE

 StoreType.PICKLE

Returns:

Type Description
_R

Object loaded from cache OR object created with callback and stored to cache.
Source code in magic_storage/mixins/_cache_if_missing.py 
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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
def cache_if_missing(
    self,
    uid: str,
    callback: Callable[[], _R],
    store_type: StoreType = StoreType.PICKLE,
) -> _R:
    """Store and return object if not present in cache, otherwise load from
    cache and return.

    In case of load failure object cache is recreated.

    Parameters
    ----------
    uid : str
        Object identifier used to find object in cache.
    callback : Callable[[], _R]
        Callback function which can create new object if object is not found in cache
    store_type : StoreType, optional
        Determines how object should be stored in cache, by default StoreType.PICKLE

    Returns
    -------
    _R
        Object loaded from cache OR object created with callback and stored to cache.
    """
    uid = make_uid(uid)

    if self.is_available(uid):
        logging.debug(f"'{uid}' is available and will be loaded.")
        try:
            return self.load_as(store_type, uid=uid)  # type: ignore
        except Exception as e:
            logging.exception(e)
        logging.warning(
            f"Failed to load '{uid}' due to loading error. Cache will be recreated."
        )

    else:
        logging.debug(
            f"Resource '{uid}' is NOT available thus will be created."
        )
    # If cache is not present OR if cache load failed
    item = callback()
    self.store_as(store_type, uid=uid, item=item)
    return item  # type: ignore
Last update: August 5, 2022
Created: August 5, 2022