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
|
# Copyright (C) 2009-2011 AG Projects. See LICENSE for details.
#
"""Miscellaneous utility descriptors"""
__all__ = ['ThreadLocal', 'WriteOnceAttribute', 'classproperty', 'isdescriptor']
import weakref
from threading import local
class ThreadLocal(object):
"""Descriptor that allows objects to have thread specific attributes"""
thread_local = local()
def __init__(self, type, *args, **kw):
self.type = type
self.args = args
self.kw = kw
def __get__(self, obj, objtype):
if obj is None:
return self
obj_id = id(obj)
self_id = id(self)
try:
return self.thread_local.__dict__[(obj_id, self_id)][0]
except KeyError:
instance = self.type(*self.args, **self.kw)
thread_dict = self.thread_local.__dict__
ref = weakref.ref(obj, lambda weak_ref: thread_dict.pop((obj_id, self_id)))
thread_dict[(obj_id, self_id)] = (instance, ref)
return instance
def __set__(self, obj, value):
obj_id = id(obj)
self_id = id(self)
thread_dict = self.thread_local.__dict__
if (obj_id, self_id) in thread_dict:
ref = thread_dict[(obj_id, self_id)][1]
else:
ref = weakref.ref(obj, lambda weak_ref: thread_dict.pop((obj_id, self_id)))
thread_dict[(obj_id, self_id)] = (value, ref)
def __delete__(self, obj):
raise AttributeError("attribute cannot be deleted")
class WriteOnceAttribute(object):
"""
Descriptor that allows objects to have write once attributes.
It should be noted that the descriptor only enforces this when directly
accessing the object's attribute. It is still possible to modify/delete
such an attribute by messing around with the descriptor's internal data.
"""
def __init__(self):
self.values = {}
def __get__(self, obj, type):
if obj is None:
return self
try:
return self.values[id(obj)][0]
except KeyError:
raise AttributeError("attribute is not set")
def __set__(self, obj, value):
obj_id = id(obj)
if obj_id in self.values:
raise AttributeError("attribute is read-only")
self.values[obj_id] = (value, weakref.ref(obj, lambda weak_ref: self.values.pop(obj_id)))
def __delete__(self, obj):
raise AttributeError("attribute cannot be deleted")
def classproperty(function):
"""A class level read only property"""
class Descriptor(object):
def __get__(self, obj, type):
return function(type)
def __set__(self, obj, value):
raise AttributeError("read-only attribute cannot be set")
def __delete__(self, obj):
raise AttributeError("read-only attribute cannot be deleted")
return Descriptor()
def isdescriptor(object):
"""Test if `object' is a descriptor"""
return bool(set(('__get__', '__set__', '__delete__')).intersection(dir(object)))
|
