Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
Locate all setuptools entry points by the name 'plexiglas plugins' and initialize them.
Any third-party library may register an entry point by adding the following to their setup.py::
entry_points = {
'plexiglas.plugins': [
'plugin_name = mylib.mymodule:initialize_func',
],
},
`plugin_name` can be anything, and is only used to display the name
of the plugin at initialization time.
`initialize_func` is optional, but will be invoked if callable.
"""
group = 'plexiglas.plugins'
entry_points = entrypoints.get_group_all(group=group)
for ep in entry_points:
try:
log.info('Loading plugin %s', ep.name)
init_func = ep.load()
if callable(init_func):
init_func()
except Exception:
log.exception("Error initializing plugin %s." % ep)
def load_extensions():
"""Return list of Kibitzr CLI extensions"""
return [
point.load()
for point in entrypoints.get_group_all("kibitzr.cli")
]
def from_entrypoints(cls):
"""
Load all kernel providers advertised by entry points.
Kernel providers should use the "jupyter_kernel_mgmt.kernel_type_providers"
entry point group.
Returns an instance of KernelFinder.
"""
providers = []
for ep in entrypoints.get_group_all('jupyter_kernel_mgmt.kernel_type_providers'):
try:
provider = ep.load()() # Load and instantiate
except Exception:
log.error('Error loading kernel provider', exc_info=True)
else:
providers.append(provider)
return cls(providers)
def _load_entrypoint_plugins(self):
LOG.info('Loading entry-points for "%s".', self.namespace)
for entry_point in entrypoints.get_group_all(self.namespace):
if entry_point.name == "per-file-ignores":
LOG.warning(
"flake8-per-file-ignores plugin is incompatible with "
"flake8>=3.7 (which implements per-file-ignores itself)."
)
continue
self._load_plugin_from_entrypoint(entry_point)
def register_entrypoints(self):
"""Register tracking stores provided by other packages"""
for entrypoint in entrypoints.get_group_all(self.group_name):
try:
self.register(entrypoint.name, entrypoint.load())
except (AttributeError, ImportError) as exc:
warnings.warn(
'Failure attempting to register store for scheme "{}": {}'.format(
entrypoint.name, str(exc)
),
stacklevel=2
)
def register_entrypoints(self):
# Register artifact repositories provided by other packages
for entrypoint in entrypoints.get_group_all("mlflow.artifact_repository"):
try:
self.register(entrypoint.name, entrypoint.load())
except (AttributeError, ImportError) as exc:
warnings.warn(
'Failure attempting to register artifact repository for scheme "{}": {}'.format(
entrypoint.name, str(exc)
),
stacklevel=2
)
def _load_filters():
"""
Locate all entry points by the name 'pmxbot_filters', each of
which should refer to a callable(channel, msg) that must return
True for the message not to be excluded.
"""
group = 'pmxbot_filters'
eps = entrypoints.get_group_all(group=group)
return [ep.load() for ep in eps]
def register_entrypoints(self):
"""Register tracking stores provided by other packages"""
for entrypoint in entrypoints.get_group_all("mlflow.run_context_provider"):
try:
self.register(entrypoint.load())
except (AttributeError, ImportError) as exc:
warnings.warn(
'Failure attempting to register context provider "{}": {}'.format(
entrypoint.name, str(exc)
),
stacklevel=2
)
def load_extensions():
"""Return list of fetcher promoters defined in Kibitzr extensions"""
return [
point.load()
for point in entrypoints.get_group_all("kibitzr.fetcher")
]
def register_entry_points(self):
# Load handlers provided by other packages
for entrypoint in entrypoints.get_group_all("papermill.io"):
self.register(entrypoint.name, entrypoint.load())