mirror
/
amqtt
kopia lustrzana https://github.com/Yakifo/amqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Replace pkg_resources with importlib.metadata 

pkg_resources is removed in python 3.12 and replaced
by importlib.metadata.
pull/161/head
Martin Pilros 2024-10-09 11:27:03 +02:00
rodzic 08028ebef8
commit f2b0f995d0
1 zmienionych plików z 16 dodań i 11 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

27
amqtt/plugins/manager.py
Wyświetl plik

@ -4,7 +4,7 @@
__all__ = ["get_plugin_manager", "BaseContext", "PluginManager"] __all__ = ["get_plugin_manager", "BaseContext", "PluginManager"]
import pkg_resources from importlib.metadata import EntryPoint, entry_points
import logging import logging
import asyncio import asyncio
import copy import copy
@ -59,24 +59,29 @@ class PluginManager:
def _load_plugins(self, namespace): def _load_plugins(self, namespace):
self.logger.debug("Loading plugins for namespace %s" % namespace) self.logger.debug("Loading plugins for namespace %s" % namespace)
for ep in pkg_resources.iter_entry_points(group=namespace): if hasattr(entry_points(), "select"):
plugin = self._load_plugin(ep) ep = entry_points().select(group=namespace)
self._plugins.append(plugin) elif namespace in entry_points():
self.logger.debug(" Plugin %s ready" % ep.name) ep = entry_points()[namespace]
else:
ep = []
def _load_plugin(self, ep: pkg_resources.EntryPoint): for item in ep:
plugin = self._load_plugin(item)
self._plugins.append(plugin)
self.logger.debug(" Plugin %s ready" % item.name)
def _load_plugin(self, ep: EntryPoint):
try: try:
self.logger.debug(" Loading plugin %s" % ep) self.logger.debug(" Loading plugin %s" % str(ep))
plugin = ep.load(require=True) plugin = ep.load()
self.logger.debug(" Initializing plugin %s" % ep) self.logger.debug(" Initializing plugin %s" % str(ep))
plugin_context = copy.copy(self.app_context) plugin_context = copy.copy(self.app_context)
plugin_context.logger = self.logger.getChild(ep.name) plugin_context.logger = self.logger.getChild(ep.name)
obj = plugin(plugin_context) obj = plugin(plugin_context)
return Plugin(ep.name, ep, obj) return Plugin(ep.name, ep, obj)
except ImportError as ie: except ImportError as ie:
self.logger.warning(f"Plugin {ep!r} import failed: {ie}") self.logger.warning(f"Plugin {ep!r} import failed: {ie}")
except pkg_resources.UnknownExtra as ue:
self.logger.warning(f"Plugin {ep!r} dependencies resolution failed: {ue}")
def get_plugin(self, name): def get_plugin(self, name):
""" """