Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: modernize importlib #11

Open
wants to merge 1 commit into
base: dev
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 2 additions & 18 deletions hivemind_plugin_manager/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@
from typing import Optional, Dict, Any, Union, Type

from ovos_utils.log import LOG

from importlib.metadata import entry_points
from hivemind_plugin_manager.database import AbstractDB, AbstractRemoteDB
from hivemind_plugin_manager.protocols import AgentProtocol, BinaryDataHandlerProtocol, NetworkProtocol

Expand Down Expand Up @@ -89,22 +89,6 @@ def create(cls, plugin_name: str,
return plugin(config=config, hm_protocol=hm_protocol, agent_protocol=agent_protocol)


def _iter_entrypoints(plug_type: Optional[str]):
"""
Return an iterator containing all entrypoints of the requested type
@param plug_type: entrypoint name to load
@return: iterator of all entrypoints
"""
try:
from importlib_metadata import entry_points
for entry_point in entry_points(group=plug_type):
yield entry_point
except ImportError:
import pkg_resources
for entry_point in pkg_resources.iter_entry_points(plug_type):
yield entry_point


def find_plugins(plug_type: HiveMindPluginTypes = None) -> dict:
"""
Finds all plugins matching specific entrypoint type.
Expand All @@ -123,7 +107,7 @@ def find_plugins(plug_type: HiveMindPluginTypes = None) -> dict:
else:
plugs = plug_type
for plug in plugs:
for entry_point in _iter_entrypoints(plug):
for entry_point in entry_points(group=plug):
try:
entrypoints[entry_point.name] = entry_point.load()
if entry_point.name not in entrypoints:
Expand Down
Loading