"""
functions for dynamically loading objects from entry point object references
"""
import functools
import os
import typing
import warnings
import pkg_resources
__all__ = (
'LazyDescriptor',
'from_string',
'from_environment',
'from_file',
'from_working_set',
)
[docs]def from_string(object_reference: str):
"""
resolves an object as specificed by an environment variable
.. testcode:: string
''' xyzzy.py '''
class Breakfast(object):
toast = 'burnt'
.. testcode:: string
:hide:
import sys
import argparse
sys.modules['xyzzy'] = argparse.Namespace(Breakfast=Breakfast)
.. doctest:: string
>>> from fictive.patterns.dynamic_load import from_string
>>> from_string(object_reference='xyzzy:Breakfast.toast')
'burnt'
:param str object_reference:
an object reference string according to
https://packaging.python.org/specifications/entry-points/#data-model
:raises ValueError:
the object reference contains one or more syntax errors
:raises ImportError: the entry point could not be loaded
"""
entry_point_source = f"from_string = {object_reference}"
entry_point = pkg_resources.EntryPoint.parse(entry_point_source)
return entry_point.resolve()
[docs]class LazyDescriptor(object):
"""
uses `pkg_resources` resolution to provide lazy-loading / late-binding
.. testcode:: descriptor
''' mypackage.py '''
class MyClass(object):
THIS_CLASS = MyClass # NameError
.. testoutput:: descriptor
Traceback (most recent call last):
...
NameError: name 'MyClass' is not defined
.. testcode:: descriptor
''' mypackage.py '''
from fictive.patterns.dynamic_load import LazyDescriptor
class MyClass(object):
THIS_CLASS = LazyDescriptor('mypackage:MyClass')
.. testcode:: descriptor
:hide:
import sys
import argparse
sys.modules['mypackage'] = argparse.Namespace(MyClass=MyClass)
.. doctest:: descriptor
>>> MyClass.THIS_CLASS is MyClass
True
"""
[docs] def __init__(self, object_reference, cached=True):
self.object_reference = object_reference
self.cached = cached
@functools.cached_property
def resolved(self):
"""
resolve (and cache) the object reference
"""
return from_string(self.object_reference)
def __get__(self, obj, type=None): # pylint: disable=redefined-builtin
try:
return self.resolved
finally:
if not self.cached:
del self.resolved
[docs]def from_environment(env_key: str):
"""
resolves an object as specificed by an environment variable
.. testcode:: environment
''' xyzzy.py '''
class Breakfast(object):
eggs = 'scrambled'
.. testcode:: environment
:hide:
import sys
import argparse
sys.modules['xyzzy'] = argparse.Namespace(Breakfast=Breakfast)
.. doctest:: environment
>>> import os
>>> object_reference = f"xyzzy:Breakfast.eggs"
>>> os.environ['FICTIVE_DYNAMIC_EGGS'] = object_reference
>>> from fictive.patterns.dynamic_load import from_environment
>>> from_environment(env_key='FICTIVE_DYNAMIC_EGGS')
'scrambled'
:param str env_key:
the key for an environment variable that contains an object
reference string according to
https://packaging.python.org/specifications/entry-points/#data-model
:raises KeyError: the environment does not contain a value for the key
:raises ValueError:
the object reference contains one or more syntax errors
:raises ImportError: the entry point could not be loaded
"""
object_reference = os.environ[env_key]
entry_point_source = f"from_environment = {object_reference}"
entry_point = pkg_resources.EntryPoint.parse(entry_point_source)
return entry_point.resolve()
[docs]def from_file(group: str, name: str, filename: str = 'entry_points.txt'):
"""
resolves an entry point as specified in a file
.. testsetup:: file
with open('entry_points.txt', mode='w') as entry_points_file:
entry_points_file.write('[foo.bar]\\n')
entry_points_file.write('baz = xyzzy:Breakfast.bacon\\n')
entry_points_file.write('[other.group]\\n')
entry_points_file.write('name = other:object.reference\\n')
.. testcode:: file
''' xyzzy.py '''
class Breakfast(object):
bacon = 'crispy'
.. testcode:: file
:hide:
import sys
import argparse
sys.modules['xyzzy'] = argparse.Namespace(Breakfast=Breakfast)
.. code-block:: ini
# entry_points.txt
[foo.bar]
baz = xyzzy:Breakfast.bacon
[other.group]
name = other:object.reference
.. doctest:: file
>>> group = 'foo.bar'
>>> name = 'baz'
>>> from fictive.patterns.dynamic_load import from_file
>>> from_file(group=group, name=name)
'crispy'
.. testcleanup:: file
import os
os.remove('entry_points.txt')
:param str group: the "group" for the entry point.
:param str name: the "name" for the entry point
:param str filename:
the name of an entry points according to
https://packaging.python.org/specifications/entry-points/#file-format
:raises OSError: the file could not be opened
:raises ValueError:
the entry points file contains one or more syntax errors
:raises KeyError:
the group or entry name within the group was not found in the
entry point file
:raises ImportError: the entry point could not be loaded
"""
entry_point_file = open(filename, mode='r')
lines = list(entry_point_file)
entry_point_file.close()
entry_point_map = pkg_resources.EntryPoint.parse_map(lines)
return entry_point_map[group][name].resolve()
[docs]def from_working_set(
group: str, name: str, working_set: pkg_resources.WorkingSet = None,
multiple: typing.Literal['raise', 'warn', 'silent', 'iter'] = 'raise'):
"""
resolves an entry point from a `pkg_resources.WorkingSet`
.. testcode:: workingset
''' xyzzy.py '''
class Breakfast(object):
potatoes = 'hashed'
.. code-block:: python
''' setup.py '''
from setuptools import setup
setup(
...,
entry_points=(
'foo.bar': {
'baz': 'xyzzy:Breakfast.potatoes'
}
}
)
.. testcode:: workingset
:hide:
import sys
import argparse
sys.modules['xyzzy'] = argparse.Namespace(Breakfast=Breakfast)
.. testcode:: workingset
:hide:
import pkg_resources
distribution = pkg_resources.Distribution(
'_mock_distribution_', project_name='_mock_distribution_')
entry_map = {'foo.bar': ['baz = xyzzy:Breakfast.potatoes']}
parsed = pkg_resources.EntryPoint.parse_map(entry_map, distribution)
distribution._ep_map = parsed
pkg_resources.working_set.add(distribution)
.. doctest:: workingset
>>> group = 'foo.bar'
>>> name = 'baz'
>>> from fictive.patterns.dynamic_load import from_working_set
>>> from_working_set(group=group, name=name)
'hashed'
:param str group: the "group" for the entry point.
:param str name: the "name" for the entry point
:param working_set:
the working set to search for the entry point. If not
specified, uses the default `pkg_resources.working_set`
:type working_set: `pkg_resources.WorkingSet`
:param multiple:
`pkg_resources` permits a working set to have multiple entry
points with the same group and name. In the event that
multiple entry points match the `group` and `name` values
provided, the value of the `multiple` parameter controls the
result:
- 'raise' (default):
raises a `pkg_resources.ResolutionError`
- 'warn':
a warning is issued and one of the matching entry points
is arbitrarily chosen and loaded
- 'silent':
one of the matching entry points is arbitrarily chosen and
loaded without any warning
:raises ValueError:
the contains invalid entry point syntax
:raises pkg_resources.ResolutionError:
there are multiple entry points in the specified group with
the specified name and `multiple` is :code:`'raise'`
:raises ImportError: the entry point could not be loaded
"""
working_set = working_set or pkg_resources.working_set
entry_points = list(working_set.iter_entry_points(group=group, name=name))
if len(entry_points) > 1:
error_message = (
'At most one distribution in the working set can define the'
f' "{group}":"{name}" entry point (defined in'
f' {[ep.dist.project_name for ep in entry_points]}).')
if multiple == 'raise':
raise pkg_resources.ResolutionError(error_message)
if multiple == 'warn':
warnings.warn(error_message + ' Selecting one at random...')
return entry_points[0].resolve()