diff options
Diffstat (limited to 'migrate/versioning/util')
| -rw-r--r-- | migrate/versioning/util/__init__.py | 187 | ||||
| -rw-r--r-- | migrate/versioning/util/importpath.py | 30 | ||||
| -rw-r--r-- | migrate/versioning/util/keyedinstance.py | 36 |
3 files changed, 0 insertions, 253 deletions
diff --git a/migrate/versioning/util/__init__.py b/migrate/versioning/util/__init__.py deleted file mode 100644 index 55c72c9..0000000 --- a/migrate/versioning/util/__init__.py +++ /dev/null @@ -1,187 +0,0 @@ -#!/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)) - - ep = EntryPoint.parse('x=%s' % dotted_name) - if hasattr(ep, 'resolve'): - # this is available on setuptools >= 10.2 - return ep.resolve() - else: - # this causes a DeprecationWarning on setuptools >= 11.3 - return ep.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] diff --git a/migrate/versioning/util/importpath.py b/migrate/versioning/util/importpath.py deleted file mode 100644 index 529be89..0000000 --- a/migrate/versioning/util/importpath.py +++ /dev/null @@ -1,30 +0,0 @@ -import os -import sys - -PY33 = sys.version_info >= (3, 3) - -if PY33: - from importlib import machinery -else: - from six.moves import reload_module as reload - - -def import_path(fullpath): - """ Import a file with full path specification. Allows one to - import from anywhere, something __import__ does not do. - """ - if PY33: - name = os.path.splitext(os.path.basename(fullpath))[0] - return machinery.SourceFileLoader( - name, fullpath).load_module(name) - else: - # http://zephyrfalcon.org/weblog/arch_d7_2002_08_31.html - path, filename = os.path.split(fullpath) - filename, ext = os.path.splitext(filename) - sys.path.append(path) - try: - module = __import__(filename) - reload(module) # Might be out of date during tests - return module - finally: - del sys.path[-1] diff --git a/migrate/versioning/util/keyedinstance.py b/migrate/versioning/util/keyedinstance.py deleted file mode 100644 index a692e08..0000000 --- a/migrate/versioning/util/keyedinstance.py +++ /dev/null @@ -1,36 +0,0 @@ -#!/usr/bin/env python -# -*- coding: utf-8 -*- - -class KeyedInstance(object): - """A class whose instances have a unique identifier of some sort - No two instances with the same unique ID should exist - if we try to create - a second instance, the first should be returned. - """ - - _instances = dict() - - def __new__(cls, *p, **k): - instances = cls._instances - clskey = str(cls) - if clskey not in instances: - instances[clskey] = dict() - instances = instances[clskey] - - key = cls._key(*p, **k) - if key not in instances: - instances[key] = super(KeyedInstance, cls).__new__(cls) - return instances[key] - - @classmethod - def _key(cls, *p, **k): - """Given a unique identifier, return a dictionary key - This should be overridden by child classes, to specify which parameters - should determine an object's uniqueness - """ - raise NotImplementedError() - - @classmethod - def clear(cls): - # Allow cls.clear() as well as uniqueInstance.clear(cls) - if str(cls) in cls._instances: - del cls._instances[str(cls)] |