Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/celery/utils/objects.py : 24%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# -*- coding: utf-8 -*-
2"""Object related utilities, including introspection, etc."""
3from __future__ import absolute_import, unicode_literals
5from functools import reduce
7__all__ = ('Bunch', 'FallbackContext', 'getitem_property', 'mro_lookup')
10class Bunch(object):
11 """Object that enables you to modify attributes."""
13 def __init__(self, **kwargs):
14 self.__dict__.update(kwargs)
17def mro_lookup(cls, attr, stop=None, monkey_patched=None):
18 """Return the first node by MRO order that defines an attribute.
20 Arguments:
21 cls (Any): Child class to traverse.
22 attr (str): Name of attribute to find.
23 stop (Set[Any]): A set of types that if reached will stop
24 the search.
25 monkey_patched (Sequence): Use one of the stop classes
26 if the attributes module origin isn't in this list.
27 Used to detect monkey patched attributes.
29 Returns:
30 Any: The attribute value, or :const:`None` if not found.
31 """
32 stop = set() if not stop else stop
33 monkey_patched = [] if not monkey_patched else monkey_patched
34 for node in cls.mro():
35 if node in stop:
36 try:
37 value = node.__dict__[attr]
38 module_origin = value.__module__
39 except (AttributeError, KeyError):
40 pass
41 else:
42 if module_origin not in monkey_patched:
43 return node
44 return
45 if attr in node.__dict__:
46 return node
49class FallbackContext(object):
50 """Context workaround.
52 The built-in ``@contextmanager`` utility does not work well
53 when wrapping other contexts, as the traceback is wrong when
54 the wrapped context raises.
56 This solves this problem and can be used instead of ``@contextmanager``
57 in this example::
59 @contextmanager
60 def connection_or_default_connection(connection=None):
61 if connection:
62 # user already has a connection, shouldn't close
63 # after use
64 yield connection
65 else:
66 # must've new connection, and also close the connection
67 # after the block returns
68 with create_new_connection() as connection:
69 yield connection
71 This wrapper can be used instead for the above like this::
73 def connection_or_default_connection(connection=None):
74 return FallbackContext(connection, create_new_connection)
75 """
77 def __init__(self, provided, fallback, *fb_args, **fb_kwargs):
78 self.provided = provided
79 self.fallback = fallback
80 self.fb_args = fb_args
81 self.fb_kwargs = fb_kwargs
82 self._context = None
84 def __enter__(self):
85 if self.provided is not None:
86 return self.provided
87 context = self._context = self.fallback(
88 *self.fb_args, **self.fb_kwargs
89 ).__enter__()
90 return context
92 def __exit__(self, *exc_info):
93 if self._context is not None:
94 return self._context.__exit__(*exc_info)
97class getitem_property(object):
98 """Attribute -> dict key descriptor.
100 The target object must support ``__getitem__``,
101 and optionally ``__setitem__``.
103 Example:
104 >>> from collections import defaultdict
106 >>> class Me(dict):
107 ... deep = defaultdict(dict)
108 ...
109 ... foo = _getitem_property('foo')
110 ... deep_thing = _getitem_property('deep.thing')
113 >>> me = Me()
114 >>> me.foo
115 None
117 >>> me.foo = 10
118 >>> me.foo
119 10
120 >>> me['foo']
121 10
123 >>> me.deep_thing = 42
124 >>> me.deep_thing
125 42
126 >>> me.deep
127 defaultdict(<type 'dict'>, {'thing': 42})
128 """
130 def __init__(self, keypath, doc=None):
131 path, _, self.key = keypath.rpartition('.')
132 self.path = path.split('.') if path else None
133 self.__doc__ = doc
135 def _path(self, obj):
136 return (reduce(lambda d, k: d[k], [obj] + self.path) if self.path
137 else obj)
139 def __get__(self, obj, type=None):
140 if obj is None:
141 return type
142 return self._path(obj).get(self.key)
144 def __set__(self, obj, value):
145 self._path(obj)[self.key] = value