Source code for turses.meta

# -*- coding: utf-8 -*-

This module contains abstract classes and decorators.

import logging
from abc import ABCMeta, abstractmethod, abstractproperty
from functools import wraps
from threading import Thread

# - Decorators ----------------------------------------------------------------

[docs]def wrap_exceptions(func): """ Augments the function arguments with the `on_error` and `on_success` keyword arguments. Executes the decorated function in a try except block and calls `on_success` (if given) if no exception was raised, otherwise calls `on_error` (if given). """ @wraps(func) def wrapper(self=None, *args, **kwargs): on_error = kwargs.pop('on_error', None) on_success = kwargs.pop('on_success', None) try: result = func(self, *args, **kwargs) except Exception, message: if callable(on_error): logging.exception(message) on_error() else: if callable(on_success): on_success() return result return wrapper
[docs]def async(func): """ Decorator for executing a function in a separate :class:`threading.Thread`. """ @wraps(func) def wrapper(*args, **kwargs): thread = Thread(target=func, args=args, kwargs=kwargs) thread.daemon = True return thread.start() return wrapper
[docs]def filter_result(func, filter_func=None): """ Decorator for filtering the output of ``func`` with ``filter_func``. """ @wraps(func) def wrapper(*args, **kwargs): result = func(*args, **kwargs) if isinstance(result, list): return [filter_func(elem) for elem in result] else: return filter_func(result) return wrapper # - Abstract classes ---------------------------------------------------------- # FIXME: Use urwid.MonitoredFocusList
[docs]class ActiveList(object): """ A list that contains an *active* element. This abstract class implements some functions but the subclasses must define the :attr:`` property, as well as :func:`turses.meta.ActiveList.is_valid_index` and :func:`turses.meta.ActiveList.activate_last`. methods. """ __metaclass__ = ABCMeta NULL_INDEX = -1 def __init__(self): self.active_index = self.NULL_INDEX @abstractproperty def active(self): pass @abstractmethod def is_valid_index(self, index): pass def activate_previous(self): """Mark as active the previous element if it exists.""" new_index = self.active_index - 1 if self.is_valid_index(new_index): self.active_index = new_index def activate_next(self): """Mark as active the next element if it exists.""" new_index = self.active_index + 1 if self.is_valid_index(new_index): self.active_index = new_index def activate_first(self): """Mark the first element as the 'active' if it exists.""" first = 0 if self.is_valid_index(first): self.active_index = first else: self.active_index = self.NULL_INDEX @abstractmethod def activate_last(self): pass
[docs]class UnsortedActiveList(ActiveList): """ A :class:`~turses.meta.ActiveList` in which the *active* element position can be shifted either to the beginning or to the end. All the methods contained in this class are abstract. """ @abstractmethod def shift_active_previous(self): """Shift the active element one position to the left.""" pass @abstractmethod def shift_active_next(self): """Shift the active element one position to the right.""" pass @abstractmethod def shift_active_beggining(self): """ Shift the active element (if any) to the beginning of the list. """ pass @abstractmethod def shift_active_end(self): """ Shift the active element (if any) to the beginning of the list. """ pass
[docs]class Updatable: """ An abstract class for making a class *updatable*. The constructor takes an update function and arguments used to update the subclasses of :class:`~turses.meta.Updatable`. When :func:`~turses.meta.Updatable.update` is executed, :func:`~turses.meta.Updatable.update_callback` is called, passing it the result. """ __metaclass__ = ABCMeta def __init__(self, update_function=None, update_function_args=None, update_function_kwargs=None,): """ `update_function` is the function used to update the class, with optional args `update_function_args` and `update_function_kwargs`. """ self.update_function = update_function if isinstance(update_function_args, tuple): self._args = list(update_function_args) elif update_function_args: self._args = [update_function_args] else: self._args = [] if update_function_kwargs: self._kwargs = dict(update_function_kwargs) else: self._kwargs = {} @wrap_exceptions def update(self, **extra_kwargs): """ Update the object. The result of `update_function` is passed to the `update_callback` function. """ if not self.update_function: return args = self._args kwargs = self._kwargs kwargs.update(extra_kwargs) result = self.update_function(*args, **kwargs) self.update_callback(result) @abstractmethod def update_callback(self, result): pass
[docs]def notify(func): """ Wrap an instance method `func`, calling the instance's `notify` method after executing `func`. """ @wraps(func) def wrapper(self, *args, **kwargs): result = func(self, *args, **kwargs) if isinstance(self, Observable): self.notify() else: raise Exception("The `notify` decorator should only be applied to " "`Observable` subclasses") return result return wrapper
[docs]class Observable: """ An implementation of the *observer* pattern. Zero or more *observers* can `subscribe` to the changes in the instances of this class. When the instance changes, it will call its `notify` method, which loops through the *observers* and calls `update()` on them. """ def __init__(self): self._observers = set([]) def subscribe(self, observer): self._observers.add(observer) def unsubscribe(self, observer): self._observers.discard(observer) def notify(self): for observer in self._observers: observer.update()
[docs]class Observer: """ An abstract class that can subscribe to updates from :class:`~turses.meta.Observable` instances. """ __metaclass__ = ABCMeta @abstractmethod def update(self): pass
Read the Docs v: latest
On Read the Docs
Project Home

Free document hosting provided by Read the Docs.