/usr/lib/python2.7/dist-packages/cached_property.py is in python-cached-property 1.3.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 65 66 67 68 69 70 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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | # -*- coding: utf-8 -*-
__author__ = 'Daniel Greenfeld'
__email__ = 'pydanny@gmail.com'
__version__ = '1.3.0'
__license__ = 'BSD'
from time import time
import threading
class cached_property(object):
"""
A property that is only computed once per instance and then replaces itself
with an ordinary attribute. Deleting the attribute resets the property.
Source: https://github.com/bottlepy/bottle/commit/fa7733e075da0d790d809aa3d2f53071897e6f76
""" # noqa
def __init__(self, func):
self.__doc__ = getattr(func, '__doc__')
self.func = func
def __get__(self, obj, cls):
if obj is None:
return self
value = obj.__dict__[self.func.__name__] = self.func(obj)
return value
class threaded_cached_property(object):
"""
A cached_property version for use in environments where multiple threads
might concurrently try to access the property.
"""
def __init__(self, func):
self.__doc__ = getattr(func, '__doc__')
self.func = func
self.lock = threading.RLock()
def __get__(self, obj, cls):
if obj is None:
return self
obj_dict = obj.__dict__
name = self.func.__name__
with self.lock:
try:
# check if the value was computed before the lock was acquired
return obj_dict[name]
except KeyError:
# if not, do the calculation and release the lock
return obj_dict.setdefault(name, self.func(obj))
class cached_property_with_ttl(object):
"""
A property that is only computed once per instance and then replaces itself
with an ordinary attribute. Setting the ttl to a number expresses how long
the property will last before being timed out.
"""
def __init__(self, ttl=None):
if callable(ttl):
func = ttl
ttl = None
else:
func = None
self.ttl = ttl
self._prepare_func(func)
def __call__(self, func):
self._prepare_func(func)
return self
def __get__(self, obj, cls):
if obj is None:
return self
now = time()
obj_dict = obj.__dict__
name = self.__name__
try:
value, last_updated = obj_dict[name]
except KeyError:
pass
else:
ttl_expired = self.ttl and self.ttl < now - last_updated
if not ttl_expired:
return value
value = self.func(obj)
obj_dict[name] = (value, now)
return value
def __delete__(self, obj):
obj.__dict__.pop(self.__name__, None)
def __set__(self, obj, value):
obj.__dict__[self.__name__] = (value, time())
def _prepare_func(self, func):
self.func = func
if func:
self.__doc__ = func.__doc__
self.__name__ = func.__name__
self.__module__ = func.__module__
# Aliases to make cached_property_with_ttl easier to use
cached_property_ttl = cached_property_with_ttl
timed_cached_property = cached_property_with_ttl
class threaded_cached_property_with_ttl(cached_property_with_ttl):
"""
A cached_property version for use in environments where multiple threads
might concurrently try to access the property.
"""
def __init__(self, ttl=None):
super(threaded_cached_property_with_ttl, self).__init__(ttl)
self.lock = threading.RLock()
def __get__(self, obj, cls):
with self.lock:
return super(threaded_cached_property_with_ttl, self).__get__(obj,
cls)
# Alias to make threaded_cached_property_with_ttl easier to use
threaded_cached_property_ttl = threaded_cached_property_with_ttl
timed_threaded_cached_property = threaded_cached_property_with_ttl
|