Pyblish in 200 lines

pyblish_in_200_lines.py

import json
import types
import inspect

# The core Pyblish package is mocked.
# These are all relevant members to processing.
pyblish = types.ModuleType("pyblish")
pyblish.api = types.ModuleType("api")
pyblish.plugin = types.ModuleType("plugin")
pyblish.plugin.process = None   # Implemented below
pyblish.plugin.Provider = None  # Implemented below
pyblish.logic = types.ModuleType("logic")
pyblish.logic.process = None    # Implemented below


# This metaclass hides certain (internal) aspects of plug-ins that
# the end-user (e.g. you) aren't intended to ever touch.
class MetaPlugin(type):
    def __init__(cls, *args, **kwargs):
        args_ = inspect.getargspec(cls.process).args
        cls.__instanceEnabled__ = "instance" in args_


class Plugin(object):
    __metaclass__ = MetaPlugin

    order = 0

    def __init__(self):
        # This illustrates how logging works and how logs are
        # stored and retrieved during processing.
        self._records = list()
        self.log = type("Logger", (object,), {})()
        self.log.info = self._records.append

    def process(self):
        # The default implementation does nothing
        # hence no need to call upon super().
        pass


# The CVEI superclasses, also does nothing but define a default order
Collector = type("Collector", (Plugin,), {"order": 0})
Validator = type("Validator", (Plugin,), {"order": 1})
Extractor = type("Extractor", (Plugin,), {"order": 2})
Integrator = type("Integrator", (Plugin,), {"order": 3})


class Provider(object):
    """The dependency-injection mechanism

    Available arguments to a plug-ins `process()` method
    are injected here and later passed to the function via `invoke()`.

    Example
        >>> def function(name):
        ...     print(name)
        ...
        >>> provider = Provider()
        >>> provider.inject("name", "Marcus")
        >>> provider.invoke(function)
        "Marcus"

    """

    def __init__(self):
        self._services = {}

    def inject(self, name, obj):
        self._services[name] = obj

    def invoke(self, func):
        args = self.args(func)
        return func(**{k: v for k, v in self._services.items()
                       if k in args})

    @classmethod
    def args(cls, func):
        return [a for a in inspect.getargspec(func)[0]
                if a not in ("self", "cls")]

pyblish.plugin.Provider = Provider


# Example plugins
class CollectInstances(Collector):
    def process(self, context):
        context.append("instance1")
        context.append("instance2")
        self.log.info("Processing context..")


class ValidateInstances(Validator):
    def process(self, instance):
        self.log.info("Processing %s" % instance)


# Ordering
plugins = [ValidateInstances, CollectInstances]
plugins = sorted(plugins, key=lambda item: item.order)


# Primary processing function.
# This function runs your plug-in and produces a `result` dictionary.
def process(plugin, context, instance):
    print("plugin.process running..")
    print("Individually processing \"%s\"" % (instance or "Context"))

    result = {
        "plugin": plugin.__name__,
        "instance": None,
        "error": None,
        "records": list(),
        "success": False
    }

    provider = pyblish.plugin.Provider()
    provider.inject("plugin", plugin)
    provider.inject("context", context)
    provider.inject("instance", instance)

    plugin = plugin()

    try:
        provider.invoke(plugin.process)
    except Exception as e:
        result["success"] = False
        result["error"] = e

    result["records"] = plugin._records

    return result

pyblish.plugin.process = process


# Logical processing function.
# This function delegates processing the appropriate plug-in/instance pair.
def process(func, plugins, context):
    print("logic.process running..")

    for plugin in plugins:
        print("Processing plug-in: %s" % plugin)

        # Run once
        if not plugin.__instanceEnabled__:
            yield func(
                plugin=plugin,
                context=context,
                instance=None)

        # Run per instance
        else:
            for instance in context:
                yield func(
                    plugin=plugin,
                    context=context,
                    instance=instance)

pyblish.logic.process = process


# Example usage
def publish():
    context = list()
    processor = pyblish.logic.process(
        func=pyblish.plugin.process,
        plugins=plugins,
        context=context
    )

    print("Starting..")
    results = list()
    for result in processor:
        results.append(result)

    print("\nFinished, printing results")
    print(json.dumps(results, indent=4))

if __name__ == "__main__":
    publish()
    # Starting..
    # logic.process running..
    # Processing plug-in: <class '__main__.CollectInstances'>
    # plugin.process running..
    # Individually processing "Context"
    # Processing plug-in: <class '__main__.ValidateInstances'>
    # plugin.process running..
    # Individually processing "instance1"
    # plugin.process running..
    # Individually processing "instance2"

    # Finished, printing results
    # [
    #     {
    #         "instance": null,
    #         "error": null,
    #         ...