/usr/lib/python2.7/dist-packages/migrate/versioning/util/__init__.py is in python-migrate 0.10.0-3ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | #!/usr/bin/env python
# -*- coding: utf-8 -*-
""".. currentmodule:: migrate.versioning.util"""
import warnings
import logging
from decorator import decorator
from pkg_resources import EntryPoint
import six
from sqlalchemy import create_engine
from sqlalchemy.engine import Engine
from sqlalchemy.pool import StaticPool
from migrate import exceptions
from migrate.versioning.util.keyedinstance import KeyedInstance
from migrate.versioning.util.importpath import import_path
log = logging.getLogger(__name__)
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 asbool(obj):
"""Do everything to use object as bool"""
if isinstance(obj, six.string_types):
obj = obj.strip().lower()
if obj in ['true', 'yes', 'on', 'y', 't', '1']:
return True
elif obj in ['false', 'no', 'off', 'n', 'f', '0']:
return False
else:
raise ValueError("String is not true/false: %r" % obj)
if obj in (True, False):
return bool(obj)
else:
raise ValueError("String is not true/false: %r" % obj)
def guess_obj_type(obj):
"""Do everything to guess object type from string
Tries to convert to `int`, `bool` and finally returns if not succeded.
.. versionadded: 0.5.4
"""
result = None
try:
result = int(obj)
except:
pass
if result is None:
try:
result = asbool(obj)
except:
pass
if result is not None:
return result
else:
return obj
@decorator
def catch_known_errors(f, *a, **kw):
"""Decorator that catches known api errors
.. versionadded: 0.5.4
"""
try:
return f(*a, **kw)
except exceptions.PathFoundError as e:
raise exceptions.KnownError("The path %s already exists" % e.args[0])
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)
@decorator
def with_engine(f, *a, **kw):
"""Decorator for :mod:`migrate.versioning.api` functions
to safely close resources after function usage.
Passes engine parameters to :func:`construct_engine` and
resulting parameter is available as kw['engine'].
Engine is disposed after wrapped function is executed.
.. versionadded: 0.6.0
"""
url = a[0]
engine = construct_engine(url, **kw)
try:
kw['engine'] = engine
return f(*a, **kw)
finally:
if isinstance(engine, Engine) and engine is not url:
log.debug('Disposing SQLAlchemy engine %s', engine)
engine.dispose()
class Memoize(object):
"""Memoize(fn) - an instance which acts like fn but memoizes its arguments
Will only work on functions with non-mutable arguments
ActiveState Code 52201
"""
def __init__(self, fn):
self.fn = fn
self.memo = {}
def __call__(self, *args):
if args not in self.memo:
self.memo[args] = self.fn(*args)
return self.memo[args]
|