raise NotImplementedError()
+ @util.deprecated('Deprecated. Use construct_params(). '
+ '(supports Unicode key names.)')
def get_params(self, **params):
- """Use construct_params(). (supports unicode names)"""
return self.construct_params(params)
- get_params = util.deprecated()(get_params)
def construct_params(self, params):
"""Return the bind params for this compiled object.
key = self._only_mapper_zero().identity_key_from_primary_key(ident)
return self._get(key, ident)
+ @classmethod
+ @util.deprecated('Deprecated. Use sqlalchemy.orm.with_parent '
+ 'in conjunction with filter().')
def query_from_parent(cls, instance, property, **kwargs):
"""Return a new Query with criterion corresponding to a parent instance.
all extra keyword arguments are propagated to the constructor of
Query.
- deprecated. use sqlalchemy.orm.with_parent in conjunction with
- filter().
-
"""
mapper = object_mapper(instance)
prop = mapper.get_property(property, resolve_synonyms=True)
target = prop.mapper
criterion = prop.compare(operators.eq, instance, value_is_parent=True)
return Query(target, **kwargs).filter(criterion)
- query_from_parent = classmethod(util.deprecated(None, False)(query_from_parent))
@_generative()
def correlate(self, *args):
self._deleted.pop(state, None)
del state.session_id
+ @util.pending_deprecation('0.5.x', "Use session.add()")
def save(self, instance, entity_name=None):
"""Add a transient (unsaved) instance to this ``Session``.
state = _state_for_unsaved_instance(instance, entity_name)
self._save_impl(state)
self._cascade_save_or_update(state, entity_name)
- save = util.pending_deprecation('0.5.x', "Use session.add()")(save)
def _save_without_cascade(self, instance, entity_name=None):
"""Used by scoping.py to save on init without cascade."""
state = _state_for_unsaved_instance(instance, entity_name, create=True)
self._save_impl(state)
+ @util.pending_deprecation('0.5.x', "Use session.add()")
def update(self, instance, entity_name=None):
"""Bring a detached (saved) instance into this ``Session``.
raise exc.UnmappedInstanceError(instance, entity_name)
self._update_impl(state)
self._cascade_save_or_update(state, entity_name)
- update = util.pending_deprecation('0.5.x', "Use session.add()")(update)
def add(self, instance, entity_name=None):
"""Add the given instance into this ``Session``.
return self._bind is not None
- # @deprecated
+ @util.deprecated('Deprecated. Use ``metadata.bind = <engine>`` or '
+ '``metadata.bind = <url>``.')
def connect(self, bind, **kwargs):
"""Bind this MetaData to an Engine.
- Use ``metadata.bind = <engine>`` or ``metadata.bind = <url>``.
-
bind
A string, ``URL``, ``Engine`` or ``Connection`` instance. If a
string or ``URL``, will be passed to ``create_engine()`` along with
self._bind = create_engine(bind, **kwargs)
else:
self._bind = bind
- connect = util.deprecated()(connect)
def bind(self):
"""An Engine or Connection to which this MetaData is bound.
self.__engines = {}
super(ThreadLocalMetaData, self).__init__()
- # @deprecated
+ @util.deprecated('Deprecated. Use ``metadata.bind = <engine>`` or '
+ '``metadata.bind = <url>``.')
def connect(self, bind, **kwargs):
"""Bind to an Engine in the caller's thread.
- Use ``metadata.bind=<engine>`` or ``metadata.bind=<url>``.
-
bind
A string, ``URL``, ``Engine`` or ``Connection`` instance. If a
string or ``URL``, will be passed to ``create_engine()`` along with
engine = create_engine(bind, **kwargs)
bind = engine
self._bind_to(bind)
- connect = util.deprecated()(connect)
def bind(self):
"""The bound Engine or Connection for this thread.
def _decorate_with_warning(func, wtype, message, docstring_header=None):
"""Wrap a function with a warnings.warn and augmented docstring."""
- def func_with_warning(*args, **kwargs):
- warnings.warn(wtype(message), stacklevel=2)
- return func(*args, **kwargs)
+ @decorator
+ def warned(fn, *args, **kwargs):
+ warnings.warn(wtype(message), stacklevel=3)
+ return fn(*args, **kwargs)
doc = func.__doc__ is not None and func.__doc__ or ''
if docstring_header is not None:
- doc = '\n'.join((docstring_header.rstrip(), doc))
-
- func_with_warning.__doc__ = doc
- func_with_warning.__dict__.update(func.__dict__)
+ docstring_header %= dict(func=func.__name__)
+ docs = doc and doc.expandtabs().split('\n') or []
+ indent = ''
+ for line in docs[1:]:
+ text = line.lstrip()
+ if text:
+ indent = line[0:len(line) - len(text)]
+ break
+ point = min(len(docs), 1)
+ docs.insert(point, '\n' + indent + docstring_header.rstrip())
+ doc = '\n'.join(docs)
- return function_named(func_with_warning, func.__name__)
+ decorated = warned(func)
+ decorated.__doc__ = doc
+ return decorated
"assign this Table's .metadata.bind to enable implicit "
"execution.")
- @testing.uses_deprecated('//connect')
+ @testing.uses_deprecated()
def test_create_drop_bound(self):
for meta in (MetaData,ThreadLocalMetaData):
assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
# test static method
- @testing.uses_deprecated(".*query_from_parent")
+ @testing.uses_deprecated(".*Use sqlalchemy.orm.with_parent")
def go():
o = Query.query_from_parent(u1, property='orders', session=sess).all()
assert [Order(description="order 1"), Order(description="order 3"), Order(description="order 5")] == o
self.assert_compile(s, "SELECT foo, bar UNION SELECT foo, bar UNION (SELECT foo, bar UNION SELECT foo, bar)")
- @testing.uses_deprecated('//get_params')
+ @testing.uses_deprecated()
def test_binds(self):
for (
stmt,
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
