Source code for warg.decorators.caching.property_caching

#!/usr/bin/env python3

__author__ = "Christian Heider Lindbjerg"
__doc__ = r"""
          TODO: Will be part of functools in the future!

           Created on 16/02/2020
           """

# TODO: Will be part of functools in the future!

__all__ = ["cached_property"]

import logging
from typing import Any

_logger = logging.getLogger(__name__)
# cached_property = property

################################################################################
### cached_property() - computed once per instance, cached as attribute
################################################################################
try:
    from _thread import RLock
except:
    pass

_NOT_FOUND = object()




class cached_property:
    """description"""



    def __init__(self, func):
        self.func = func
        self.attrname = None
        self.__doc__ = func.__doc__
        self.lock = RLock()


    def __set_name__(self, owner, name):
        if self.attrname is None:
            self.attrname = name
        elif name != self.attrname:
            raise TypeError(
                "Cannot assign the same cached_property to two different names "
                f"({self.attrname!r} and {name!r})."
            )

    def __get__(self, instance, owner=None) -> Any:
        if instance is None:
            return self
        if self.attrname is None:
            raise TypeError("Cannot use cached_property instance without calling __set_name__ on it.")
        try:
            cache = instance.__dict__
        except AttributeError:  # not all objects have __dict__ (e.g. class defines slots)
            msg = (
                f"No '__dict__' attribute on {type(instance).__name__!r} "
                f"instance to cache {self.attrname!r} property."
            )
            raise TypeError(msg) from None
        val = cache.get(self.attrname, _NOT_FOUND)
        if val is _NOT_FOUND:
            with self.lock:
                # check if another thread filled cache while we awaited lock
                val = cache.get(self.attrname, _NOT_FOUND)
                if val is _NOT_FOUND:
                    val = self.func(instance)
                    try:
                        cache[self.attrname] = val
                    except TypeError:
                        msg = (
                            f"The '__dict__' attribute on {type(instance).__name__!r} instance "
                            f"does not support item assignment for caching {self.attrname!r} property."
                        )
                        raise TypeError(msg) from None
        return val