- implemented code cleanup from [ticket:1152] but not including using the decorators module
connection. The memo will be stored in ``connection.info[key]``.
"""
- def decorate(fn):
- spec = inspect.getargspec(fn)
- assert len(spec[0]) == 2
- assert spec[0][1] == 'connection'
- assert spec[1:3] == (None, None)
-
- def decorated(self, connection):
- try:
- return connection.info[key]
- except KeyError:
- connection.info[key] = val = fn(self, connection)
- return val
+ @util.decorator
+ def decorated(fn, self, connection):
+ try:
+ return connection.info[key]
+ except KeyError:
+ connection.info[key] = val = fn(self, connection)
+ return val
- return util.function_named(decorated, fn.__name__)
- return decorate
+ return decorated
class AttributeImpl(object):
"""internal implementation for instrumented attributes."""
- def __init__(self, class_, key, callable_, class_manager, trackparent=False, extension=None, compare_function=None, active_history=False, **kwargs):
+ def __init__(self, class_, key,
+ callable_, class_manager, trackparent=False, extension=None,
+ compare_function=None, active_history=False, **kwargs):
"""Construct an AttributeImpl.
\class_
uses_objects = False
- def __init__(self, class_, key, callable_, class_manager, copy_function=None, compare_function=None, **kwargs):
- super(ScalarAttributeImpl, self).__init__(class_, key, callable_, class_manager, compare_function=compare_function, **kwargs)
+ def __init__(self, class_, key, callable_,
+ class_manager, copy_function=None,
+ compare_function=None, **kwargs):
+ super(ScalarAttributeImpl, self).__init__(class_, key, callable_,
+ class_manager, compare_function=compare_function, **kwargs)
class_manager.mutable_attributes.add(key)
if copy_function is None:
raise sa_exc.ArgumentError("MutableScalarAttributeImpl requires a copy function")
accepts_scalar_loader = False
uses_objects = True
- def __init__(self, class_, key, callable_, class_manager, trackparent=False, extension=None, copy_function=None, compare_function=None, **kwargs):
+ def __init__(self, class_, key, callable_, class_manager,
+ trackparent=False, extension=None, copy_function=None,
+ compare_function=None, **kwargs):
super(ScalarObjectAttributeImpl, self).__init__(class_, key,
callable_, class_manager, trackparent=trackparent, extension=extension,
compare_function=compare_function, **kwargs)
accepts_scalar_loader = False
uses_objects = True
- def __init__(self, class_, key, callable_, class_manager, typecallable=None, trackparent=False, extension=None, copy_function=None, compare_function=None, **kwargs):
+ def __init__(self, class_, key, callable_, class_manager,
+ typecallable=None, trackparent=False, extension=None,
+ copy_function=None, compare_function=None, **kwargs):
super(CollectionAttributeImpl, self).__init__(class_,
key, callable_, class_manager, trackparent=trackparent,
extension=extension, compare_function=compare_function, **kwargs)
manager.instantiable = False
manager.unregister()
-def register_attribute(class_, key, uselist, useobject, callable_=None, proxy_property=None, mutable_scalars=False, impl_class=None, **kwargs):
+def register_attribute(class_, key, uselist, useobject,
+ callable_=None, proxy_property=None,
+ mutable_scalars=False, impl_class=None, **kwargs):
manager = manager_of_class(class_)
if manager.is_instrumented(key):
return
return sync.source_changes(uowcommit, state, self.parent, self.prop.synchronize_pairs)
class MapperStub(object):
- """Pose as a Mapper representing the association table in a
- many-to-many join, when performing a ``flush()``.
+ """Represent a many-to-many dependency within a flush
+ context.
+
+ The UOWTransaction corresponds dependencies to mappers.
+ MapperStub takes the place of the "association table"
+ so that a depedendency can be corresponded to it.
- The ``Task`` objects in the objectstore module treat it just like
- any other ``Mapper``, but in fact it only serves as a dependency
- placeholder for the many-to-many update task.
"""
-
__metaclass__ = util.ArgSingleton
def __init__(self, parent, mapper, key):
return from_obj
@property
- @util.cache_decorator
+ @util.memoize
def _with_polymorphic_mappers(self):
if not self.with_polymorphic:
return [self]
return self.__mappers_from_spec(*self.with_polymorphic)
@property
- @util.cache_decorator
+ @util.memoize
def _with_polymorphic_selectable(self):
if not self.with_polymorphic:
return self.mapped_table
self.version_id_col = self.inherits.version_id_col
for mapper in self.iterate_to_root():
- util.reset_cached(mapper, '_equivalent_columns')
+ util.reset_memoized(mapper, '_equivalent_columns')
if self.order_by is False and not self.concrete and self.inherits.order_by is not False:
self.order_by = self.inherits.order_by
self.__log("Identified primary key columns: " + str(primary_key))
@property
- @util.cache_decorator
+ @util.memoize
def _get_clause(self):
"""create a "get clause" based on the primary key. this is used
by query.get() and many-to-one lazyloads to load this item
return sql.and_(*[k==v for (k, v) in params]), dict(params)
@property
- @util.cache_decorator
+ @util.memoize
def _equivalent_columns(self):
"""Create a map of all *equivalent* columns, based on
the determination of column pairs that are equated to
return value
class ColumnComparator(PropComparator):
+ @util.memoize
def __clause_element__(self):
return self.prop.columns[0]._annotate({"parententity": self.mapper})
- __clause_element__ = util.cache_decorator(__clause_element__)
def operate(self, op, *other, **kwargs):
return op(self.__clause_element__(), *other, **kwargs)
def _generative(*assertions):
"""Mark a method as generative."""
+
+ @util.decorator
def generate(fn, *args, **kw):
self = args[0]._clone()
fn_name = fn.func_name
assertion(self, fn_name)
fn(self, *args[1:], **kw)
return self
- return util.decorator(generate)
+ return generate
class Query(object):
"""Encapsulates the object-fetching operations provided by Mappers."""
return 'defaultdict(%s, %s)' % (self.default_factory,
dict.__repr__(self))
-
def to_list(x, default=None):
if x is None:
return default
else:
return x
+def to_set(x):
+ if x is None:
+ return set()
+ if not isinstance(x, set):
+ return set(to_list(x))
+ else:
+ return x
+
+
try:
from functools import update_wrapper
except ImportError:
spec = inspect.getargspec(fn)
assert spec[1], 'Decorated function does not accept *args'
- meta = format_argspec_plus(spec)
- meta['name'] = fn.func_name
- meta['varg'] = spec[1]
- scratch = list(spec)
- scratch[1] = '(%s[0])' % scratch[1]
- meta['unpacked_pos'] = format_argspec_plus(scratch)['apply_pos']
-
def _deprecate():
if list_deprecation:
if list_deprecation == 'pending':
warning_type = exc.SAPendingDeprecationWarning
else:
warning_type = exc.SADeprecationWarning
- msg = (
- "%s%s now accepts multiple %s arguments as a "
- "variable argument list. Supplying %s as a single "
- "list is deprecated and support will be removed "
- "in a future release." % (
- fn.func_name,
- inspect.formatargspec(*spec),
- spec[1], spec[1]))
- warnings.warn(msg, warning_type, stacklevel=3)
-
- code = "\n".join((
- "def %(name)s%(args)s:",
- " if len(%(varg)s) == 1 and isinstance(%(varg)s[0], list):",
- " _deprecate()",
- " return fn%(unpacked_pos)s",
- " else:",
- " return fn%(apply_pos)s")) % meta
-
- env = locals().copy()
- exec code in env
- decorated = env[fn.func_name]
- decorated.func_defaults = fn.func_defaults
- update_wrapper(decorated, fn)
- decorated.generated_src = code
- return decorated
+ msg = (
+ "%s%s now accepts multiple %s arguments as a "
+ "variable argument list. Supplying %s as a single "
+ "list is deprecated and support will be removed "
+ "in a future release." % (
+ fn.func_name,
+ inspect.formatargspec(*spec),
+ spec[1], spec[1]))
+ warnings.warn(msg, warning_type, stacklevel=3)
+
+ def go(fn, *args, **kw):
+ if isinstance(args[-1], list):
+ _deprecate()
+ return fn(*(list(args[0:-1]) + args[-1]), **kw)
+ else:
+ return fn(*args, **kw)
+
+ return decorator(go)(fn)
+
return decorate
def unique_symbols(used, *bases):
return update_wrapper(decorated, fn)
return update_wrapper(decorate, target)
-def to_set(x):
- if x is None:
- return set()
- if not isinstance(x, set):
- return set(to_list(x))
- else:
- return x
-
-def to_ascii(x):
- """Convert Unicode or a string with unknown encoding into ASCII."""
-
- if isinstance(x, str):
- return x.encode('string_escape')
- elif isinstance(x, unicode):
- return x.encode('unicode_escape')
- else:
- raise TypeError
if sys.version_info >= (2, 5):
def decode_slice(slc):
pass
setattr(into_cls, method, env[method])
-class SimpleProperty(object):
- """A *default* property accessor."""
-
- def __init__(self, key):
- self.key = key
-
- def __set__(self, obj, value):
- setattr(obj, self.key, value)
-
- def __delete__(self, obj):
- delattr(obj, self.key)
-
- def __get__(self, obj, owner):
- if obj is None:
- return self
- else:
- return getattr(obj, self.key)
-
-
-class NotImplProperty(object):
- """a property that raises ``NotImplementedError``."""
-
- def __init__(self, doc):
- self.__doc__ = doc
-
- def __set__(self, obj, value):
- raise NotImplementedError()
-
- def __delete__(self, obj):
- raise NotImplementedError()
-
- def __get__(self, obj, owner):
- if obj is None:
- return self
- else:
- raise NotImplementedError()
class OrderedProperties(object):
"""An object that maintains the order in which attributes are set upon it.
class UniqueAppender(object):
- """Only adds items to a collection once.
+ """Appends items to a collection ensuring uniqueness.
Additional appends() of the same object are ignored. Membership is
determined by identity (``is a``) not equality (``==``).
fn.func_defaults, fn.func_closure)
return fn
-def cache_decorator(func):
+@decorator
+def memoize(fn, self):
"""apply caching to the return value of a function."""
- name = '_cached_' + func.__name__
+ name = '_cached_' + fn.__name__
- def do_with_cache(self, *args, **kwargs):
- try:
- return getattr(self, name)
- except AttributeError:
- value = func(self, *args, **kwargs)
- setattr(self, name, value)
- return value
- return do_with_cache
+ try:
+ return getattr(self, name)
+ except AttributeError:
+ value = fn(self)
+ setattr(self, name, value)
+ return value
-def reset_cached(instance, name):
+def reset_memoized(instance, name):
try:
delattr(instance, '_cached_' + name)
except AttributeError:
del self.by_id[key]
except (KeyError, AttributeError): # pragma: no cover
pass # pragma: no cover
- if sys.version_info < (2, 4): # pragma: no cover
- def _ref(self, object):
- oid = id(object)
- return weakref.ref(object, lambda wr: self._cleanup(wr, oid))
- else:
- class _keyed_weakref(weakref.ref):
- def __init__(self, object, callback):
- weakref.ref.__init__(self, object, callback)
- self.key = id(object)
-
- def _ref(self, object):
- return self._keyed_weakref(object, self._cleanup)
+
+ class _keyed_weakref(weakref.ref):
+ def __init__(self, object, callback):
+ weakref.ref.__init__(self, object, callback)
+ self.key = id(object)
+
+ def _ref(self, object):
+ return self._keyed_weakref(object, self._cleanup)
def warn(msg):
raise Exception("this exception should be stated as a warning")
sess.expunge = bad_expunge
- try:
- Foo(_sa_session=sess)
- assert False
- except Exception, e:
- assert isinstance(e, sa.exc.SAWarning), e
+ self.assertRaises(sa.exc.SAWarning, Foo, _sa_session=sess)
@testing.resolve_artifact_names
def test_constructor_exc_2(self):
mapper(Foo, users)
mapper(Bar, addresses)
- try:
- Foo(x=5)
- assert False
- except TypeError:
- assert True
-
- try:
- Bar(x=5)
- assert False
- except TypeError:
- assert True
+ self.assertRaises(TypeError, Foo, x=5)
+ self.assertRaises(TypeError, Bar, x=5)
@testing.resolve_artifact_names
def test_props(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
