You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
data-transport/transport/plugins/__init__.py

129 lines
4.4 KiB
Python

"""
The functions within are designed to load external files and apply functions against the data
The plugins are applied as
- post-processing if we are reading data
- and pre-processing if we are writing data
The plugin will use a decorator to identify meaningful functions
@TODO: This should work in tandem with loggin (otherwise we don't have visibility into what is going on)
"""
import importlib as IL
import importlib.util
import sys
import os
class plugin :
"""
Implementing function decorator for data-transport plugins (post-pre)-processing
"""
def __init__(self,**_args):
"""
:name name of the plugin
:mode restrict to reader/writer
:about tell what the function is about
"""
self._name = _args['name']
self._about = _args['about']
self._mode = _args['mode'] if 'mode' in _args else 'rw'
def __call__(self,pointer):
def wrapper(_args):
return pointer(_args)
#
# @TODO:
# add attributes to the wrapper object
#
setattr(wrapper,'transport',True)
setattr(wrapper,'name',self._name)
setattr(wrapper,'mode',self._mode)
setattr(wrapper,'about',self._about)
return wrapper
class PluginLoader :
"""
This class is intended to load a plugin and make it available and assess the quality of the developed plugin
"""
def __init__(self,**_args):
"""
:path location of the plugin (should be a single file)
:_names of functions to load
"""
_names = _args['names'] if 'names' in _args else None
path = _args['path'] if 'path' in _args else None
self._names = _names if type(_names) == list else [_names]
self._modules = {}
self._names = []
if path and os.path.exists(path) and _names:
for _name in self._names :
spec = importlib.util.spec_from_file_location('private', path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) #--loads it into sys.modules
if hasattr(module,_name) :
if self.isplugin(module,_name) :
self._modules[_name] = getattr(module,_name)
else:
print ([f'Found {_name}', 'not plugin'])
else:
#
# @TODO: We should log this somewhere some how
print (['skipping ',_name, hasattr(module,_name)])
pass
else:
#
# Initialization is empty
self._names = []
pass
def set(self,_pointer) :
"""
This function will set a pointer to the list of modules to be called
This should be used within the context of using the framework as a library
"""
_name = _pointer.__name__
self._modules[_name] = _pointer
self._names.append(_name)
def isplugin(self,module,name):
"""
This function determines if a module is a recognized plugin
:module module object loaded from importlib
:name name of the functiion of interest
"""
p = type(getattr(module,name)).__name__ =='function'
q = hasattr(getattr(module,name),'transport')
#
# @TODO: add a generated key, and more indepth validation
return p and q
def has(self,_name):
"""
This will determine if the module name is loaded or not
"""
return _name in self._modules
def ratio (self):
"""
how many modules loaded vs unloaded given the list of names
"""
_n = len(self._names)
return len(set(self._modules.keys()) & set (self._names)) / _n
def apply(self,_data):
for _name in self._modules :
_pointer = self._modules[_name]
#
# @TODO: add exception handling
_data = _pointer(_data)
return _data
# def apply(self,_data,_name):
# """
# This function applies an external module function against the data.
# The responsibility is on the plugin to properly return data, thus responsibility is offloaded
# """
# try:
# _pointer = self._modules[_name]
# _data = _pointer(_data)
# except Exception as e:
# pass
# return _data