Source code for ftrack_api.plugin

# :coding: utf-8
# :copyright: Copyright (c) 2014 ftrack

from __future__ import absolute_import

import logging
import os
import uuid
import traceback

import importlib.util
import importlib.machinery


import inspect


[docs]def load_source(modname, filename): # https://docs.python.org/3/whatsnew/3.12.html#imp loader = importlib.machinery.SourceFileLoader(modname, filename) module = importlib.util.module_from_spec( importlib.util.spec_from_file_location(modname, filename, loader=loader) ) loader.exec_module(module) return module
[docs]def discover(paths, positional_arguments=None, keyword_arguments=None): """Find and load plugins in search *paths*. Each discovered module should implement a register function that accepts *positional_arguments* and *keyword_arguments* as \*args and \*\*kwargs respectively. If a register function does not accept variable arguments, then attempt to only pass accepted arguments to the function by inspecting its signature. """ logger = logging.getLogger(__name__ + ".discover") if positional_arguments is None: positional_arguments = [] if keyword_arguments is None: keyword_arguments = {} for path in paths: # Ignore empty paths that could resolve to current directory. path = path.strip() if not path: continue for base, directories, filenames in os.walk(path): for filename in filenames: name, extension = os.path.splitext(filename) if extension != ".py": continue module_path = os.path.join(base, filename) unique_name = uuid.uuid4().hex try: module = load_source(unique_name, module_path) except Exception as error: logger.warning( 'Failed to load plugin from "{0}": {1}'.format( module_path, error ) ) logger.debug(traceback.format_exc()) continue try: module.register except AttributeError: logger.warning( "Failed to load plugin that did not define a " '"register" function at the module level: {0}'.format( module_path ) ) else: # Attempt to only pass arguments that are accepted by the # register function. specification = inspect.getfullargspec(module.register) selected_positional_arguments = positional_arguments selected_keyword_arguments = keyword_arguments if not specification.varargs and len(positional_arguments) > len( specification.args ): logger.warning( "Culling passed arguments to match register " "function signature." ) selected_positional_arguments = positional_arguments[ len(specification.args) : ] selected_keyword_arguments = {} elif not specification.varkw: # Remove arguments that have been passed as positionals. remainder = specification.args[len(positional_arguments) :] # Determine remaining available keyword arguments. defined_keyword_arguments = [] if specification.defaults: defined_keyword_arguments = specification.args[ -len(specification.defaults) : ] remaining_keyword_arguments = set( [ keyword_argument for keyword_argument in defined_keyword_arguments if keyword_argument in remainder ] ) if not set(keyword_arguments.keys()).issubset( remaining_keyword_arguments ): logger.warning( "Culling passed arguments to match register " "function signature." ) selected_keyword_arguments = { key: value for key, value in list(keyword_arguments.items()) if key in remaining_keyword_arguments } module.register( *selected_positional_arguments, **selected_keyword_arguments )