habe und hatte schon häufiger das Problem eine Anwendung schreiben zu müssen, die Plugins zulässt/benötigt.
Habe daher eine minimale Pluginimplementierung geschrieben! Hoffe ihr habt reichlich Verbesserungsvorschläge! (Wußte irgendwie nicht in welche Rubrik das hier gehört - aber als kleines Projekt kann man es allemal betrachten.)
Hier das Modul ``plugrunner`` mit dessen Hilfe Plugins von bestimmten Verzeichnissen (Packages) geladen werden können:
Code: Alles auswählen
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
``plugrunner`` - Pluginimport vereinfachen
==========================================
Voraussetzungen
---------------
Ein Plugin ist gueltig wenn ...
1) ... es ein gueltiges Pythonpackage ist und
2) ... es sich in einem gueltigen Pythonpackage
(Verzeichnis) befindet und
3) ... wenn es in seiner ``__init__.py`` - Datei
eine Klasse namens ``Plugin`` implementiert,
die eine Subklasse der hier erstellten Klasse
``PluginBase`` ist.
Funktionsweise
--------------
Schritt 0: Pimporter - Objekt erzeugen
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Der Pluginimporter ``Pimporter`` uebernimmt die Aufgabe das
anzugebende Pluginverzeichnis - gespeichert im Attribut
``path`` - nach Plugins zu durchsuchen. Ihm muss der Pfad
uebergeben werden, an dem er nach den moeglichen Plugins
suchen soll ``Pimporter(pathtoplugins)``
Schritt 1: Suche nach Packages
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Grundvoraussetzung fuer ein Plugin ist, dass es sich
um ein gueltiges Pythonpackage handeln muss. Das
Importerobjekt muss also erst mal nach gueltigen
Packages suchen: Methode ``scan_path``.
Schritt 2: Eigenschaften fuer Packages pruefen
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Findet der Importer gueltige Pythonpackages in dem
angegebenen Verzeichnis, dann kann mit ``search_for_plugins``
geprueft werden, ob sich dort gueltige Plugins befinden.
Die Methode ``search_for_plugins`` benutzt dabei die
(protected) Methode ``_package_is_a_valid_plugin_if``, die
im Standardfall immer ``True`` zurueckgibt. Will man fuer
ein gueltiges Plugin weitere Bedingungen festlegen, so sollte
man diese in der Methode ``_package_is_a_valid_plugin_if``
implementieren:
Beispielsweise sind Wordpress-Plugins nur gueltig,
wenn sie eine Programmdatei mit gleichem Namen
wie der Pluginverzeichnisname besitzen. Dies
waere eine weitere Bedingung, die man in der Methode
``_package_is_a_valid_plugin_if`` pruefen koennte.
Schritt 3: Importieren der moeglichen Plugins
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Ist die Suche positiv verlaufen, dann kann man versuchen,
die gefundenen, potenziellen Plugins mit ``import_plugins``
zu importieren.
Dafuer gelten die Regeln, so wie sie fuer das Pythonmodul
``import_module()`` festgelegt sind (siehe Pythondoku zur
`importlib <https://docs.python.org/3/library/importlib.html>`_)
Schritt 4: Plugins initialisieren
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
In der Methode ``init_plugins`` wird jetzt versucht, die
gefundenen, potenziellen Plugins zu starten. Dies funktioniert
nur, wenn es eine Pluginklasse passenden Namens gibt (siehe
Argument ``pluginclass`` in ``init_plugins``) und wenn diese
Klasse die hier implementierte Klasse ``PluginBase`` erweitert.
Trifft all dies zu, dann wird versucht eine Instanz der Klasse
zu erzeugen, dazu ist eine Argumentreferenz ``appobject``
noetig, die sozusagen das Bindeglied zur aufrufenden Anwendung
ist.
Das zweite Argument ist ein Zeiger auf ein ``Pinfo`` - Objekt.
Dieses wurde vom ``Pimporter`` - Objekt angelegt und beinhaltet
eine ganze Reihe von Informationen ueber das zu oeffnende Plugin.
Zugriff der Applikation auf die Plugins
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Die Plugins selbst sind in der Lage auf die Anwendung ueber
die in `Schritt 4: Plugins initialisieren`_ beschriebene
Argumentreferenz ``appobject`` zuzugreifen (Attribut ``app``
des ``Plugin`` - Objekts)
Umgekehrt kann die Anwendung auf das Plugin ueber das
``Pimporter`` - Objekt zugreifen: Dazu hat dieser die
Methode ``get_pluginobject(pluginname)`` implementiert.
Funktionsgleich damit ist der Shortcut ``__getitem__``,
also sind Zugriffe wie ``pimporter[pluginname]`` moeglich::
>>> class App(object):
... def __init__(self):
... self.pathtoplugins = "./plugins"
... self.plugins = None
... self.actualpluginobject = None
...
>>> app = App()
>>> #
>>> # load plugins
>>> pimporter = Pimporter(app.pathtoplugins)
>>> #
>>> # Step 1
>>> pimporter.scan_path()
>>> #
>>> # Step 2
>>> pimporter.search_for_plugins()
>>> #
>>> # Step 3
>>> pimporter.import_plugins("plugins")
>>> #
>>> # Step 4
>>> pimporter.init_plugins(app)
>>> #
>>> # save plugin environment
>>> app.plugins = pimporter.plugins
>>> #
>>> # get special plugin (object)
>>> specialpluginA = pimporter["specialplugin"]
>>> #
>>> # or alternatively
>>> specialpluginB = pimporter.get_pluginobject("specialplugin")
>>> app.actualpluginobject = specialpluginB
>>> #
>>> # access to the pluginobject, for example the
>>> # attribut ``data`` if there is one
>>> if hasattr(app.actualpluginobject, "data"):
... print(app.actualpluginobject.data)
"""
import logging
import os
from importlib import import_module
Logger = logging.getLogger("plugrunner")
class Pinfo(object):
"""
A plugininfo object hold collected information about a plugin.
"""
def __init__(self, name, pluginpath):
self._mro_info_ = self.__class__.__name__
self.name = name
self.path = pluginpath
# documenting the start process
self.moduleobject = None
self.classobject = None
self.pluginobject = None
# if all went right -> True
self.started = False
# if there are config data for the plugin
self.configfile = ""
self.config = None
Logger.debug("<%s> init done", self._mro_info_)
class Pimporter(object):
def __init__(self, pathtoplugins):
self._mro_info_ = self.__class__.__name__
self.path = pathtoplugins
# a directory is a package if there is a __init__.py file
# all valid packages are stored in the ``packages`` dict
# (key := name of the package, value := pathtothepackage)
self.packages = {}
# a package could be a plugin if the plugincheck is 'True'.
# ``plugins`` is a dict:
# key := pluginname
# value := ``Pinfo`` object
self.plugins = {}
Logger.debug("<%s> init done", self._mro_info_)
# 1) scan plugindirectory for packages as potential plugins
def scan_path(self):
Logger.debug("<%s> scan_path starts [path: %s]",
self._mro_info_, self.path)
self.packages = self.list_packages(self.path)
Logger.debug("<%s> scan_path: %s packages found",
self._mro_info_, len(self.packages))
def _package_is_a_valid_plugin_if(self, package):
"""This method checks, if a package is a valid plugin.
Implement your plugin conditions here.
If your plugins need some special data or files - here's
the place to check it
"""
Logger.debug("<%s> no special data for the plugin %s necessary",
self._mro_info_, package)
return True
# 2) which packages are valid plugins!
# uses ``_package_is_a_valid_plugin`` which has to be
# implemented in a subclass
def search_for_plugins(self):
"""look for valid plugins
"""
for name, package in list(self.packages.items()):
if self._package_is_a_valid_plugin_if(package):
self.plugins[name] = Pinfo(name, package)
Logger.debug("<%s> search_for_plugins: %s plugins found",
self._mro_info_, len(self.plugins))
# 3) import each plugin
def import_plugins(self, importstring, anchorpackage=None):
"""sets the ``Pinfo.module`` attribute, if the import is
successfull.
:param importstring: {str} like 'myproject.lib'
has to fit the directory structure in the actual project
:return:
"""
if importstring[-1] == ".":
importstring = importstring[:-1]
for name, pinfo in list(self.plugins.items()):
try:
# import_module('import xy', basemodule)
# if import is a relative import the second arg is needed
importname = "{}.{}".format(importstring, pinfo.name)
pinfo.moduleobject = import_module(importname, anchorpackage)
Logger.debug("<%s> import_plugins: %s plugin imported",
self._mro_info_, name)
except ImportError as e:
Logger.warning("couldn't import '%s' [%s]", pinfo.name, e)
pinfo.moduleobject = None
except TypeError as e:
Logger.warning("couldn't import '%s' [%s]", pinfo.name, e)
pinfo.moduleobject = None
# 4) initialize all plugins
def init_plugins(self, appobject, pluginclass='Plugin', *args, **kwargs):
"""initialize the plugins.
:param appobject: {parent object}
:param pluginclass: {str} name of tha plugin class in the pluginpackage
:param args: {list}
:param kwargs: {dict}
"""
for name, pinfo in list(self.plugins.items()):
if pinfo.moduleobject is not None:
try:
pinfo.classobject = \
getattr(pinfo.moduleobject, pluginclass)
Logger.debug(
"<%s> init_plugins: found PluginBase of %s",
self._mro_info_, name)
pinfo.pluginobject = \
pinfo.classobject(appobject, pinfo, *args, **kwargs)
Logger.debug(
"<%s> init_plugins: PluginBase object of %s created",
self._mro_info_, name)
except Exception as e:
Logger.warning("no entry point for plugin '%s' [%s]",
pinfo.name, e)
# Step 1) to 4)
def run(self, app, importstring, pluginclass='Plugin', *args, **kwargs):
"""runs all 4 methods at once.
:param app: {app object} look at ``init_plugins``
:param importstring: {str} look at ``import_plugins``
:param pluginclass: {str} look at ``init_plugins``
:param args: {list} look at ``init_plugins``
:param kwargs: {dict0 look at ``init_plugins``
"""
self.scan_path()
self.search_for_plugins()
self.import_plugins(importstring)
self.init_plugins(app, pluginclass, *args, **kwargs)
@staticmethod
def package_check(pathtopackage):
"""check, if ``pathtoplugin`` is valid!
:param pathtopackage: {str}
full path to a directory to check if that represens a package
:return: {str}
no package: empty string
else: full path to the package
"""
op = os.path
if op.exists(pathtopackage) and op.isdir(pathtopackage):
initfile = op.join(pathtopackage, "__init__.py")
if op.exists(initfile) and op.isfile(initfile):
return pathtopackage
return ""
@staticmethod
def list_packages(pathtopackages):
"""list all *valid* packages in ``pathtopackages``!
:param pathtopackages: {str}
path to the directory to look for packages.
:return: {dict}
key := package name
value := path to the package
"""
packages = {}
for pack in os.listdir(pathtopackages):
packagepath = os.path.join(pathtopackages, pack)
if Pimporter.package_check(packagepath):
packages.update({pack: packagepath})
return packages
# shorthand(s) for the plugin object
def get_pluginobject(self, pluginname):
if pluginname in self.plugins:
return self.plugins[pluginname].pluginobject
def __getitem__(self, item):
return self.get_pluginobject(item)
###
# valid plugins implement in the package file __init__.py
# a class named Plugin which has to be a subclass of the
# ``PluginBase`` below.
#
class PluginBase(object):
"""Ein Plugin wird vom System als gueltiges Plugin erkannt,
wenn die Packagedatei ``__init__.py`` eine Klasse ``Plugin``
beinhaltet, die von ``PluginBase`` abgeleitet ist.
class diagramm for ``PluginBase``::
+-----------------------------------------------+
| PluginBase |
| ============================================= |
| app {Application object} |
| info {Pinfo object} |
+-----------------------------------------------+
| Methods |
| ---------- |
| __init__(app, info=()) |
| _read_config() |
| run_plugin() to implement in subclass |
+-----------------------------------------------+
"""
def __init__(self, app, pinfoobject):
"""
:raises: RuntimeError
:param app: {app object}
:param pinfoobject: {Pinfo object}
"""
self._mro_info_ = self.__class__.__name__
Logger.debug("<%s> init starts [app='%s']", self._mro_info_, app)
self.app = app
if not isinstance(pinfoobject, Pinfo):
errtxt = "pinfoobject must be a Pinfo object! "
errtxt += "Got {}".format(type(pinfoobject).__name__)
raise RuntimeError(errtxt)
self.info = pinfoobject
# if a configfile is present than load/read that file
if pinfoobject.configfile:
self._read_config()
Logger.debug("<%s> init done", self._mro_info_)
def _read_config(self):
Logger.debug("<%s> no config to read", self._mro_info_)
Code: Alles auswählen
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# the file test.py
'''
lokale Test-Struktur::
.
├── __init__.py
├── plg
│ ├── __init__.py
│ └── pluginbase (<- Mein Plugin, das erkannt werden soll!)
│ ├── __init__.py
│ └── pluginbasesubmodule.py
└── test.py
'''
import logging
import os
from plugrunner import Pimporter
logging.basicConfig(
level=logging.INFO,
name="root",
propagate=1)
class App(object):
def __init__(self):
self.pluginpackage = "plg"
self.systempath = os.path.abspath(".")
self.pathtoplugins = os.path.join(self.systempath, self.pluginpackage)
self.pluginimporter = None
self.info = ""
def load(self):
self._load_plugins_at_once()
# self._load_plugins()
def _load_plugins_at_once(self):
self.pluginimporter = Pimporter(self.pathtoplugins)
self.pluginimporter.run(self, importstring=self.pluginpackage)
def _load_plugins(self):
self.pluginimporter = Pimporter(app.pathtoplugins)
self.pluginimporter.scan_path()
self.pluginimporter.search_for_plugins()
self.pluginimporter.import_plugins(importstring=self.pluginpackage)
self.pluginimporter.init_plugins(app)
def print_info(self):
print("APP-INFO: {}".format(self.info))
app = App()
print()
app.load()
connect2pluginbase = app.pluginimporter.get_pluginobject("pluginbase")
connect2pluginbase.run_plugin()
connect2pluginbase.run_plugin_again("from test_pluginrunner.py")
Code: Alles auswählen
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# the file ./plg/pluginbase/__init__.py
import logging
from plugrunner import PluginBase
from .pluginbasesubmodule import say_hello
Logger = logging.getLogger("plg.pluginbase")
class Plugin(PluginBase):
def __init__(self, app, pinfoobject):
"""
:raises: RuntimeError
:param app: {app object}
:param pinfoobject: {Pinfo object}
"""
PluginBase.__init__(self, app, pinfoobject)
self._mro_info_ = self.__class__.__name__
Logger.debug("<%s> init done", self._mro_info_)
# for testing
def run_plugin(self):
self.app.info = "Plugin '{}': ".format(self.info.name)
self.app.info += "run_plugin"
self.app.print_info()
Logger.debug("<%s> run_plugin done", self._mro_info_)
def run_plugin_again(self, addtxt=""):
self.app.info = "Plugin '{}': ".format(self.info.name)
self.app.info += "run_plugin_again: " + say_hello()
self.app.print_info()
Logger.debug("<%s> run_plugin_again done", self._mro_info_)
Code: Alles auswählen
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# ./plg/pluginbase/pluginbasemodule.py
def say_hello(app):
return "i am the pluginbasesubmodule"