| 22 | |
| 23 | |
| 24 | class Context(collections.abc.Mapping, metaclass=ContextMeta): |
| 25 | |
| 26 | def __init__(self): |
| 27 | self._data = immutables.Map() |
| 28 | self._prev_context = None |
| 29 | |
| 30 | def run(self, callable, *args, **kwargs): |
| 31 | if self._prev_context is not None: |
| 32 | raise RuntimeError( |
| 33 | 'cannot enter context: {} is already entered'.format(self)) |
| 34 | |
| 35 | self._prev_context = _get_context() |
| 36 | try: |
| 37 | _set_context(self) |
| 38 | return callable(*args, **kwargs) |
| 39 | finally: |
| 40 | _set_context(self._prev_context) |
| 41 | self._prev_context = None |
| 42 | |
| 43 | def copy(self): |
| 44 | new = Context() |
| 45 | new._data = self._data |
| 46 | return new |
| 47 | |
| 48 | def __getitem__(self, var): |
| 49 | if not isinstance(var, ContextVar): |
| 50 | raise TypeError( |
| 51 | "a ContextVar key was expected, got {!r}".format(var)) |
| 52 | return self._data[var] |
| 53 | |
| 54 | def __contains__(self, var): |
| 55 | if not isinstance(var, ContextVar): |
| 56 | raise TypeError( |
| 57 | "a ContextVar key was expected, got {!r}".format(var)) |
| 58 | return var in self._data |
| 59 | |
| 60 | def __len__(self): |
| 61 | return len(self._data) |
| 62 | |
| 63 | def __iter__(self): |
| 64 | return iter(self._data) |
| 65 | |
| 66 | |
| 67 | class ContextVarMeta(type): |
no outgoing calls
no test coverage detected
searching dependent graphs…