Source code for xotl.tools.context

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# ---------------------------------------------------------------------
# Copyright (c) Merchise Autrement [~º/~] and Contributors
# All rights reserved.
#
# This is free software; you can do what the LICENCE file allows you to.
#

"""A context manager for execution context flags."""

from xotl.tools.tasking import local
from xotl.tools.future.collections import StackedDict, Mapping

__all__ = ("Context", "context", "NullContext")


class LocalData(local):
    """Thread-local data for contexts."""

    def __init__(self):
        super().__init__()
        self.contexts = {}


_data = LocalData()


class MetaContext(type(StackedDict)):  # type: ignore
    def __len__(self):
        return len(_data.contexts)

    def __iter__(self):
        return iter(_data.contexts)

    def __getitem__(self, name):
        return _data.contexts.get(name, _null_context)

    def __contains__(self, name):
        """Basic support for the 'A in context' idiom."""
        return bool(self[name])


[docs]class Context(StackedDict, metaclass=MetaContext): """An execution context manager with parameters (or flags). Use as:: >>> SOME_CONTEXT = object() >>> from xotl.tools.context import context >>> with context(SOME_CONTEXT): ... if context[SOME_CONTEXT]: ... print('In context SOME_CONTEXT') In context SOME_CONTEXT Note the difference creating the context and checking it: for entering a context you should use ``context(name)`` for testing whether some piece of code is being executed inside a context you should use ``context[name]``; you may also use the syntax `name in context`. When an existing context is re-enter, the former one is reused. Nevertheless, the data stored in each context is local to each level. For example:: >>> with context('A', b=1) as a1: ... with context('A', b=2) as a2: ... print(a1 is a2) ... print(a2['b']) ... print(a1['b']) True 2 1 For data access, a mapping interface is provided for all contexts. If a data slot is deleted at some level, upper level is used to read values. Each new written value is stored in current level without affecting upper levels. For example:: >>> with context('A', b=1) as a1: ... with context('A', b=2) as a2: ... del a2['b'] ... print(a2['b']) 1 It is an error to *reuse* a context directly like in:: >>> with context('A', b=1) as a1: # doctest: +ELLIPSIS ... with a1: ... pass Traceback (most recent call last): ... RuntimeError: Entering the same context level twice! ... """ __slots__ = ("name", "count") def __new__(cls, name, **data): self = cls[name] if not self: # if self is _null_context: self = super().__new__(cls) super(Context, self).__init__() self.name = name self.count = 0 # TODO: Redefine all event management return self(**data)
[docs] @classmethod def from_dicts(cls, ctx, overrides=None, defaults=None): """Creates a context introducing both defaults and overrides. This combines both the standard constructor and `from_defaults`:meth:. If the same key appears in both `overrides` and `defaults`, ignore the default. """ if not overrides: overrides = {} if not defaults: defaults = {} current = cls[ctx] current_attrs = dict(current) if current else {} attrs = dict(defaults, **current_attrs) attrs.update(overrides) return cls(ctx, **attrs)
[docs] @classmethod def from_defaults(cls, ctx, **defaults): """Creates context `ctx` introducing only new keys given in `defaults`. The normal behavior when you enter a new level in the context is to override the values with the new one. Example: >>> with context.from_defaults('A', a=1): ... with context.from_defaults('A', a=2, b=1) as c: ... assert c['a'] == 1 """ return cls.from_dicts(ctx, defaults=defaults)
def __init__(self, *args, **kwargs): """Must be defined empty for `__new__` parameters compatibility. Using generic parameters definition allow any redefinition of this class can use this `__init__`. """ def __call__(self, **data): """Allow re-enter in a new level to an already assigned context.""" self.push_level(**data) return self def __nonzero__(self): return bool(self.count) __bool__ = __nonzero__ def __enter__(self): if self.count == 0: _data.contexts[self.name] = self if self.count + 1 == self.level: self.count += 1 return self else: msg = "Entering the same context level twice! -- c(%s, %d, %d)" raise RuntimeError(msg % (self.name, self.count, self.level)) def __exit__(self, exc_type, exc_value, traceback): self.count -= 1 if self.count == 0: del _data.contexts[self.name] self.pop_level() return False
# A simple alias for Context context = Context class NullContext(Mapping): """Singleton context to be used (returned) as default when no one is defined. """ __slots__ = () instance = None name = "" def __new__(cls): if cls.instance is None: cls.instance = super().__new__(cls) return cls.instance def __len__(self): return 0 def __iter__(self): return iter(()) def __getitem__(self, key): raise KeyError(key) def __nonzero__(self): return False __bool__ = __nonzero__ def __enter__(self): return _null_context def __exit__(self, exc_type, exc_value, traceback): return False def get(self, name, default=None): return default @property def level(self): return 0 _null_context = NullContext()