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
python类parse()的实例源码
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)
