Skip to content

Instantly share code, notes, and snippets.

What would you like to do?
Pyblish in 200 lines
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._records.append
def process(self):
# The default implementation does nothing
# hence no need to call upon super().
# 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()`.
>>> def function(name):
... print(name)
>>> provider = Provider()
>>> provider.inject("name", "Marcus")
>>> provider.invoke(function)
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})
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("instance2")"Processing context..")
class ValidateInstances(Validator):
def process(self, instance):"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()
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(
# Run per instance
for instance in context:
yield func(
pyblish.logic.process = process
# Example usage
def publish():
context = list()
processor = pyblish.logic.process(
results = list()
for result in processor:
print("\nFinished, printing results")
print(json.dumps(results, indent=4))
if __name__ == "__main__":
# 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,
# ...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
You can’t perform that action at this time.