aria/extension.py

# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""
Mechanism for registering and loading ARIA extensions.
"""

# pylint: disable=no-self-use

from .utils import collections


class _Registrar(object):

    def __init__(self, registry):
        if not isinstance(registry, (dict, list)):
            raise RuntimeError('Unsupported registry type')
        self._registry = registry

    def register(self, function):
        result = function()
        if isinstance(self._registry, dict):
            for key in result:
                if key in self._registry:
                    raise RuntimeError('Re-definition of {0} in {1}'.format(key, function.__name__))
            self._registry.update(result)
        elif isinstance(self._registry, list):
            if not isinstance(result, (list, tuple, set)):
                result = [result]
            self._registry += list(result)
        else:
            raise RuntimeError('Illegal state')

    def __call__(self):
        return self._registry


def _registrar(function):
    function._registrar_function = True
    return function


class _ExtensionRegistration(object):
    """
    Base class for extension class decorators.
    """

    def __init__(self):
        self._registrars = {}
        self._registered_classes = []
        for attr, value in vars(self.__class__).items():
            try:
                is_registrar_function = value._registrar_function
            except AttributeError:
                is_registrar_function = False
            if is_registrar_function:
                registrar = _Registrar(registry=getattr(self, attr)())
                setattr(self, attr, registrar)
                self._registrars[attr] = registrar

    def __call__(self, cls):
        self._registered_classes.append(cls)
        return cls

    def init(self):
        """
        Initialize all registrars by calling all registered functions.
        """
        registered_instances = [cls() for cls in self._registered_classes]
        for name, registrar in self._registrars.items():
            for instance in registered_instances:
                registrating_function = getattr(instance, name, None)
                if registrating_function:
                    registrar.register(registrating_function)


class _ParserExtensionRegistration(_ExtensionRegistration):
    """
    Parser extensions class decorator.
    """

    @_registrar
    def presenter_class(self):
        """
        Presentation class registration.

        Implementing functions can return a single class or a list/tuple of classes.
        """
        return []

    @_registrar
    def specification_package(self):
        """
        Specification package registration.

        Implementing functions can return a package name or a list/tuple of names.
        """
        return []

    @_registrar
    def specification_url(self):
        """
        Specification URL registration.

        Implementing functions should return a dictionary from names to URLs.
        """
        return {}

    @_registrar
    def uri_loader_prefix(self):
        """
        URI loader prefix registration.

        Implementing functions can return a single prefix or a list/tuple of prefixes.
        """
        return collections.StrictList(value_class=basestring)

parser = _ParserExtensionRegistration()


class _ProcessExecutorExtensionRegistration(_ExtensionRegistration):
    """
    Process executor extension class decorator.
    """

    @_registrar
    def decorate(self):
        """
        The operation function executed by the process executor will be decorated with the function
        returned from ``decorate()``.
        """
        return []

process_executor = _ProcessExecutorExtensionRegistration()


def init():
    """
    Initialize all registrars by calling all registered functions.
    """
    parser.init()
    process_executor.init()