dispatcher.py

#!/usr/bin/env python3.5
# coding: utf-8

from abc import *
from collections import ChainMap, OrderedDict

import re
from operator import eq


class DispatcherMeta(ABCMeta):
    """
    Metaclas for dispatchers. Inherits from `abc.ABCMeta` to prevent users
    from instatiating classes which do not provide an implementation for
    the static method `match(value, key) -> bool`.
    """

    def __new__(mcs, name, bases, attrs, key_type=None, factory=dict):
        callbacks = ChainMap(factory())

        for base in bases:
            if isinstance(base, DispatcherMeta):
                callbacks.maps.extend(base.__callbacks__.maps)

        attrs['__callbacks__'] = callbacks
        return super().__new__(mcs, name, bases, attrs)

    def __init__(cls, name, bases, attrs, key_type=None, factory=dict):
        super().__init__(name, bases, attrs)
        cls.key_type = key_type
        cls.factory = factory

    def get_callback_and_args(cls, value, *args, **kwds):
        """
        Returns the first callback matching `value`.
        Raises a KeyError if none is matched, or the
        default value if provided.
        """

        for key, callback in cls.__callbacks__.items():
            m = cls.match(key, value)

            if isinstance(m, bool):
                if m:
                    return callback, (value,), {}

            elif isinstance(m, tuple):
                args, kwds = m
                return callback, args, kwds

            elif m is not None:
                raise TypeError('Expected match to return a tuple or a bool '
                                'instance, got {!r}.'.format(type(m)))

        # Raises a KeyError if no default value is provided
        if not (args or kwds):
            raise KeyError(value)
        elif len(args) == 1 and not kwds:
            # Extract default value from positional arguments
            default = args
        elif len(kwds) == 1 and not args:
            # Extract default value from keyword arguments
            try:
                default = kwds['default']
            except KeyError:
                kwd = list(kwds)[0]
                raise TypeError('{}.get_callback got an unexpected keyword '
                                'argument {!r}'.format(cls.__name__, kwd))
        else:
            raise TypeError('{}.get_callback only accepts a single positional '
                            'argument and only the default keyword argument.'
                            .format(cls.__name__))

        return default, (), {}


    def on(cls, key):
        """
        Registers a callback with a given key.
        """

        if cls.key_type is not None and not isinstance(key, cls.key_type):
            raise TypeError('Expected key to be an instance of {}, got instead '
                            '{} object.'.format(cls.key_type, key))

        def _decorator_wrapper(callback):
            cls.__callbacks__[key] = callback
            return callback

        return _decorator_wrapper




class BaseDispatcher(metaclass=DispatcherMeta):
    """
    Abstract base class for dispatchers. Concrete subclasses must implement the
    static method `match(value, key) -> bool`.
    """

    @abstractstaticmethod
    def match(key, value):
        raise NotImplementedError('match static method is not implemented')

    def dispatch(self, value):
        """
        Runs a callback
        """

        callback, args, kwds = type(self).get_callback_and_args(value)
        return callback(self, *args, **kwds)


class Dispatcher(BaseDispatcher):
    """
    Matches a key on equality.
    """

    match = staticmethod(eq)


class TypeDispatcher(BaseDispatcher, factory=OrderedDict):
    """
    Matches if the provided value is an instance of the key type.
    """

    match = staticmethod(lambda key, value: isinstance(value, key))


class RegexDispatcher(BaseDispatcher, factory=OrderedDict):
    """
    Matches if the key pattern matches the whole string.
    """

    initialize_key = staticmethod(re.compile)

    @staticmethod
    def match(key, value):
        m = re.fullmatch(key, value)

        if m is not None:
            return m.groups(), m.groupdict()