Python pkg_resources.EntryPoint 模块，parse() 实例源码

def load_model(dotted_name):
    """Import module and use module-level variable".

    :param dotted_name: path to model in form of string: ``some.python.module:Class``

    .. versionchanged:: 0.5.4

    """
    if isinstance(dotted_name, basestring):
        if ':' not in dotted_name:
            # backwards compatibility
            warnings.warn('model should be in form of module.model:User '
                'and not module.model.User', exceptions.MigrateDeprecationWarning)
            dotted_name = ':'.join(dotted_name.rsplit('.', 1))
        return EntryPoint.parse('x=%s' % dotted_name).load(False)
    else:
        # Assume it's already loaded.
        return dotted_name

def load_entrypoint(cls, reference):

        # Resolve entrypoint
        expression = u'_ = {reference}'.format(reference=reference)
        try:
            entrypoint = EntryPoint.parse(expression)
        except:
            log.failure('Error parsing entrypoint "{reference}" from expression "{expression}"',
                reference=reference, expression=expression)
            raise

        # Load entrypoint
        try:
            thing = entrypoint.load(require=False)
        except:
            log.failure('Error loading entrypoint "{reference}"', reference=reference)
            raise

        return thing

def load_model(dotted_name):
    """Import module and use module-level variable".

    :param dotted_name: path to model in form of string: ``some.python.module:Class``

    .. versionchanged:: 0.5.4

    """
    if isinstance(dotted_name, six.string_types):
        if ':' not in dotted_name:
            # backwards compatibility
            warnings.warn('model should be in form of module.model:User '
                'and not module.model.User', exceptions.MigrateDeprecationWarning)
            dotted_name = ':'.join(dotted_name.rsplit('.', 1))
        return EntryPoint.parse('x=%s' % dotted_name).load(False)
    else:
        # Assume it's already loaded.
        return dotted_name

def test_dummy_implemented_manual_entrypoint(self):
        from calmjs.testing import module1
        registry = DummyModuleRegistry(__name__)
        with pretty_logging(stream=mocks.StringIO()) as s:
            registry.register_entry_point(
                EntryPoint.parse(
                    'calmjs.testing.module1 = calmjs.testing.module1')
            )
            # no dist.
            self.assertIn('manually registering entry_point', s.getvalue())
        result = registry.get_record('calmjs.testing.module1')
        self.assertEqual(result, {'calmjs.testing.module1': module1})

def test_dummy_implemented_manual_entrypoint_double_regisetr(self):
        from calmjs.testing import module1
        registry = DummyModuleRegistry(__name__)
        with pretty_logging(stream=mocks.StringIO()) as s:
            registry.register_entry_point(
                EntryPoint.parse(
                    'calmjs.testing.module1 = calmjs.testing.module1'))
            registry.register_entry_point(
                EntryPoint.parse(
                    'calmjs.testing.module1 = calmjs.testing.module1'))
            # no dist.
            self.assertIn('manually registering entry_point', s.getvalue())
        result = registry.get_record('calmjs.testing.module1')
        # just merged together.
        self.assertEqual(result, {'calmjs.testing.module1': module1})

def test_conflict_registration(self):
        # create an empty working set for a clean-slate test.
        cwd = utils.mkdtemp(self)
        mock_ws = WorkingSet([])
        registry = ArtifactRegistry('calmjs.artifacts', _working_set=mock_ws)
        # using named case for case sensitivity test.
        st = join(cwd, 'calmjs_artifacts', 'Simple.js')
        dist_ = Distribution(cwd, project_name='pkg', version='1.0')
        dist_.egg_info = cwd  # just lazy
        s1 = EntryPoint.parse('Simple.js = dummy_builder:builder1')
        s1.dist = dist_
        s2 = EntryPoint.parse('Simple.js = dummy_builder:builder2')
        s2.dist = dist_

        with pretty_logging(stream=mocks.StringIO()) as stream:
            registry.register_entry_point(s1)
            # normal registry usage shouldn't do this.
            registry.register_entry_point(s2)

        log = stream.getvalue()
        self.assertIn(
            "entry point 'Simple.js = dummy_builder:builder2' from package "
            "'pkg 1.0' will generate an artifact at '%s' but it was already "
            "registered to entry point 'Simple.js = dummy_builder:builder1'; "
            "conflicting entry point registration will be ignored." % st,
            log
        )

def test_normcase_registration(self):
        # create an empty working set for a clean-slate test.
        cwd = utils.mkdtemp(self)
        mock_ws = WorkingSet([])
        dist_ = Distribution(cwd, project_name='pkg', version='1.0')
        dist_.egg_info = cwd  # just lazy
        registry = ArtifactRegistry('calmjs.artifacts', _working_set=mock_ws)
        # case sensitive test; have to patch the normcase at artifact
        # module with the nt version
        from ntpath import normcase as nt_normcase
        utils.stub_item_attr_value(self, artifact, 'normcase', nt_normcase)
        # using named case for case sensitivity test.
        c1 = EntryPoint.parse('case.js = dummy_builder:builder1')
        c1.dist = dist_
        c2 = EntryPoint.parse('Case.js = dummy_builder:builder2')
        c2.dist = dist_
        # use the error one
        ct = join(cwd, 'calmjs_artifacts', 'Case.js')
        with pretty_logging(stream=mocks.StringIO()) as stream:
            registry.register_entry_point(c1)
            registry.register_entry_point(c2)

        log = stream.getvalue()
        self.assertIn(
            "entry point 'Case.js = dummy_builder:builder2' from package "
            "'pkg 1.0' will generate an artifact at '%s' but it was already "
            "registered to entry point 'case.js = dummy_builder:builder1'; "
            "conflicting entry point registration will be ignored." % ct,
            log
        )
        self.assertIn(
            "the file mapping error is caused by this platform's case-"
            "insensitive filename", log
        )

def test_module_registry_standard(self):
        with pretty_logging(stream=mocks.StringIO()):
            self.registry.register_entry_points([EntryPoint.parse(
                'calmjs.testing.module1 = calmjs.testing.module1')])
        self.assertEqual(sorted(
            key for key, value in self.registry.iter_records()
        ), [
            'calmjs.testing.module1',
        ])

        module1 = self.registry.get_record('calmjs.testing.module1')
        key = 'calmjs/testing/module1/hello'
        self.assertEqual(sorted(module1.keys()), [key])

def iter_entry_points(self, name):
        items = self.items.get(name, [])
        for item in items:
            entry_point = item if isinstance(
                item, EntryPoint) else EntryPoint.parse(item)
            entry_point.dist = self.dist
            yield entry_point

def run(self):
        import unittest
        import time

        # Ensure that build directory is on sys.path (py3k)
        import sys

        self.cleanup_environment()
        self.add_project_to_sys_path()

        import PyObjCTools.TestSupport as modo

        from pkg_resources import EntryPoint
        loader_ep = EntryPoint.parse("x="+self.test_loader)
        loader_class = loader_ep.load(require=False)

        try:
            meta = self.distribution.metadata
            name = meta.get_name()
            test_pkg = name + "_tests"

            time_before = time.time()
            suite = loader_class().loadTestsFromName(self.distribution.test_suite)

            runner = unittest.TextTestRunner(verbosity=self.verbosity)
            result = runner.run(suite)

            time_after = time.time()

            # Print out summary. This is a structured format that
            # should make it easy to use this information in scripts.
            summary = dict(
                count=result.testsRun,
                fails=len(result.failures),
                errors=len(result.errors),
                xfails=len(getattr(result, 'expectedFailures', [])),
                xpass=len(getattr(result, 'unexpectedSuccesses', [])),
                skip=len(getattr(result, 'skipped', [])),
                testSeconds=(time_after - time_before),
            )
            print("SUMMARY: %s"%(summary,))
            if not result.wasSuccessful():
                raise DistutilsError("some tests failed")

        finally:
            self.remove_from_sys_path()

def construct_engine(engine, **opts):
    """.. versionadded:: 0.5.4

    Constructs and returns SQLAlchemy engine.

    Currently, there are 2 ways to pass create_engine options to :mod:`migrate.versioning.api` functions:

    :param engine: connection string or a existing engine
    :param engine_dict: python dictionary of options to pass to `create_engine`
    :param engine_arg_*: keyword parameters to pass to `create_engine` (evaluated with :func:`migrate.versioning.util.guess_obj_type`)
    :type engine_dict: dict
    :type engine: string or Engine instance
    :type engine_arg_*: string
    :returns: SQLAlchemy Engine

    .. note::

        keyword parameters override ``engine_dict`` values.

    """
    if isinstance(engine, Engine):
        return engine
    elif not isinstance(engine, basestring):
        raise ValueError("you need to pass either an existing engine or a database uri")

    # get options for create_engine
    if opts.get('engine_dict') and isinstance(opts['engine_dict'], dict):
        kwargs = opts['engine_dict']
    else:
        kwargs = dict()

    # DEPRECATED: handle echo the old way
    echo = asbool(opts.get('echo', False))
    if echo:
        warnings.warn('echo=True parameter is deprecated, pass '
            'engine_arg_echo=True or engine_dict={"echo": True}',
            exceptions.MigrateDeprecationWarning)
        kwargs['echo'] = echo

    # parse keyword arguments
    for key, value in opts.iteritems():
        if key.startswith('engine_arg_'):
            kwargs[key[11:]] = guess_obj_type(value)

    log.debug('Constructing engine')
    # TODO: return create_engine(engine, poolclass=StaticPool, **kwargs)
    # seems like 0.5.x branch does not work with engine.dispose and staticpool
    return create_engine(engine, **kwargs)

def run(self):
        import unittest
        import time

        # Ensure that build directory is on sys.path (py3k)
        import sys

        self.cleanup_environment()
        self.add_project_to_sys_path()

        import PyObjCTools.TestSupport as modo

        from pkg_resources import EntryPoint
        loader_ep = EntryPoint.parse("x="+self.test_loader)
        loader_class = loader_ep.load(require=False)

        try:
            meta = self.distribution.metadata
            name = meta.get_name()
            test_pkg = name + "_tests"

            time_before = time.time()
            suite = loader_class().loadTestsFromName(self.distribution.test_suite)

            runner = unittest.TextTestRunner(verbosity=self.verbosity)
            result = runner.run(suite)

            time_after = time.time()

            # Print out summary. This is a structured format that
            # should make it easy to use this information in scripts.
            summary = dict(
                count=result.testsRun,
                fails=len(result.failures),
                errors=len(result.errors),
                xfails=len(getattr(result, 'expectedFailures', [])),
                xpass=len(getattr(result, 'unexpectedSuccesses', [])),
                skip=len(getattr(result, 'skipped', [])),
                testSeconds=(time_after - time_before),
            )
            print("SUMMARY: %s"%(summary,))
            if not result.wasSuccessful():
                raise DistutilsError("some tests failed")

        finally:
            self.remove_from_sys_path()

def construct_engine(engine, **opts):
    """.. versionadded:: 0.5.4

    Constructs and returns SQLAlchemy engine.

    Currently, there are 2 ways to pass create_engine options to :mod:`migrate.versioning.api` functions:

    :param engine: connection string or a existing engine
    :param engine_dict: python dictionary of options to pass to `create_engine`
    :param engine_arg_*: keyword parameters to pass to `create_engine` (evaluated with :func:`migrate.versioning.util.guess_obj_type`)
    :type engine_dict: dict
    :type engine: string or Engine instance
    :type engine_arg_*: string
    :returns: SQLAlchemy Engine

    .. note::

        keyword parameters override ``engine_dict`` values.

    """
    if isinstance(engine, Engine):
        return engine
    elif not isinstance(engine, six.string_types):
        raise ValueError("you need to pass either an existing engine or a database uri")

    # get options for create_engine
    if opts.get('engine_dict') and isinstance(opts['engine_dict'], dict):
        kwargs = opts['engine_dict']
    else:
        kwargs = dict()

    # DEPRECATED: handle echo the old way
    echo = asbool(opts.get('echo', False))
    if echo:
        warnings.warn('echo=True parameter is deprecated, pass '
            'engine_arg_echo=True or engine_dict={"echo": True}',
            exceptions.MigrateDeprecationWarning)
        kwargs['echo'] = echo

    # parse keyword arguments
    for key, value in six.iteritems(opts):
        if key.startswith('engine_arg_'):
            kwargs[key[11:]] = guess_obj_type(value)

    log.debug('Constructing engine')
    # TODO: return create_engine(engine, poolclass=StaticPool, **kwargs)
    # seems like 0.5.x branch does not work with engine.dispose and staticpool
    return create_engine(engine, **kwargs)