#------------------------------------------------------------------------------
# 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 __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