Source code for enaml.utils

#------------------------------------------------------------------------------
#  Copyright (c) 2011, Enthought, Inc.
#  All rights reserved.
#------------------------------------------------------------------------------
""" An amalgamation of utilities used throughout the Enaml framework.

"""
from collections import defaultdict
from functools import wraps
import logging
from random import shuffle
from string import letters, digits


[docs]def id_generator(stem): """ A unique identifier generator. For a given stem, the returned generator is guaranteed to yield consecutively increasing identifiers using a randomly ordered base 62 charset. The identifiers are only guaranteed unique for a given instance of the generator. The randomness is employed to improve the hashing characteristics of the returned identifiers. Parameters ---------- stem : str A string stem to prepend to a incrementing integer value. """ charset = list(digits + letters) shuffle(charset) charset = ''.join(charset) charsetlen = len(charset) places = [0] push = places.append enumerate_ = enumerate join = ''.join while True: yield stem + join(charset[digit] for digit in places) for idx, digit in enumerate_(places): digit += 1 if digit == charsetlen: places[idx] = 0 else: places[idx] = digit break if places[-1] == 0: push(1)
[docs]class abstractclassmethod(classmethod): """ A backport of the Python 3's abc.abstractclassmethod. """ __isabstractmethod__ = True def __init__(self, func): func.__isabstractmethod__ = True super(abstractclassmethod, self).__init__(func)
[docs]class LoopbackContext(object): """ A context manager generated by LoopbackGuard. Instances of this class manage acquiring and releasing the lock items for instances of LoopbackGuard. """ __slots__ = ('_guard', '_items')
[docs] def __init__(self, guard, items): """ Initialize a LoopbackContext Parameters ---------- guard : LoopbackGuard The loopback guard instance for which we will acquire the lock for the items. items : iterable An iterable items which will be passed to the 'acquire' method on the loopback guard. """ self._guard = guard self._items = tuple(items)
[docs] def __enter__(self): """ Acquire the guard lock on the lock items. """ self._guard.acquire(self._items)
[docs] def __exit__(self, exc_type, exc_value, traceback): """ Release the guard lock on the lock items. """ self._guard.release(self._items)
[docs]class LoopbackGuard(object): """ A guard object to protect against feedback loops. Instances of this class are used by objects to protect against loopback conditions while updating attributes. Instances of this class are callable and return a guarding context manager for the provided lock items. The guard can be tested for a locked item using the `in` keyword. """ __slots__ = ('locked_items',)
[docs] def __init__(self): """ Initialize a loopback guard. """ self.locked_items = None
[docs] def __call__(self, *items): """ Return a context manager which will guard the given items. Parameters ---------- items The items for which to acquire the guard from within the returned context manager. These items must be hashable. Returns ------- result : LoopbackContext A context manager which will acquire the guard for the provided items. """ return LoopbackContext(self, items)
[docs] def __contains__(self, item): """ Returns whether or not the given item is currently guarded. Parameters ---------- item : object The item to check for guarded state. Returns ------- result : bool True if the item is currently guarded, False otherwise. """ locked_items = self.locked_items if locked_items is not None: return item in locked_items return False
[docs] def acquire(self, items): """ Acquire the guard for the given items. This method is normally called by the LoopbackContext returned by calling this instance. User code should not typically call this method directly. It is safe to call this method multiple times for and item, provided it is paired with the same number of calls to release(...). The guard will be released when the acquired count on the item reaches zeros. Parameters ---------- items : iterable An iterable of objects for which to acquire the guard. The items must be hashable. """ locked_items = self.locked_items if locked_items is None: locked_items = self.locked_items = defaultdict(int) for item in items: locked_items[item] += 1
[docs] def release(self, items): """ Release the guard for the given lock items. This method is normally called by the LoopbackContext returned by calling this instance. User code should not normally call this method directly. It is safe to call this method multiple times for and item, provided it is paired with the same number of calls to acquire(...). The guard will be released when the acquired count on the item reaches zeros. Parameters ---------- items : iterable An iterable of objects for which to release the guard. The items must be hashable. """ locked_items = self.locked_items if locked_items is not None: for item in items: locked_items[item] -= 1 if locked_items[item] <= 0: del locked_items[item] if not locked_items: self.locked_items = None
[docs]class ObjectDict(dict): """ A dict subclass which exposes its keys as attributes. """ def __getattr__(self, name): try: return self[name] except KeyError: raise AttributeError(name) def __setattr__(self, name, value): self[name] = value
[docs]def log_exceptions(func): """ A decorator which will catch errors raised by a function and convert them into log error messages. When a decorated function raises an Exception, the return value will be None. """ @wraps(func) def closure(*args, **kwargs): try: res = func(*args, **kwargs) except Exception: # Get the logger for the wrapped function. logger = logging.getLogger(func.__module__) message = 'Exception occured in `%s`:' % func.__name__ logger.exception(message) res = None return res return closure
[docs]def make_dispatcher(prefix, logger=None): """ Create a function which will dispatch arguments to specially named handler methods on an object. Parameters ---------- prefix : str The string to prefix to all dispatch names to construct the name of the handler method. logger : logging.Logger, optional A logger to use for logging handler lookup misses. Returns ------- result : types.FunctionType A function with the signature func(obj, name, *args). Calling it is equivalent to `getattr(obj, prefix + name)(*args)` """ def dispatcher(obj, name, *args): handler = getattr(obj, prefix + name, None) if handler is not None: handler(*args) elif logger is not None: msg = "no dispatch handler found for '%s' on `%s` object" logger.warn(msg % (name, obj)) dispatcher.__name__ = prefix + '_dispatcher' return dispatcher # Backwards comatibility import. WeakMethod was moved to its own module.
from .weakmethod import WeakMethod