this is the last warning to remove.
Also fixes some mistakes I made with the new
Base20DeprecationWarning and LegacyAPIWarning classes created,
where functions in deprecations.py were still hardcoded to
RemovedIn20Warning.
Change-Id: I9a6045ac9b813fd2f9668c4bc518c46a7774c6ef
session.autoflush = autoflush
-@util.deprecated(
- "2.0",
- "The :func:`_orm.merge_result` method is superseded by the "
- ":func:`_orm.merge_frozen_result` function.",
+@util.deprecated_20(
+ ":func:`_orm.merge_result`",
+ alternative="The function as well as the method on :class:`_orm.Query` "
+ "is superseded by the :func:`_orm.merge_frozen_result` function.",
+ becomes_legacy=True,
)
@util.preload_module("sqlalchemy.orm.context")
def merge_result(query, iterator, load=True):
- """Merge a result into this :class:`.Query` object's Session."""
+ """Merge a result into the given :class:`.Query` object's Session.
+
+ See :meth:`_orm.Query.merge_result` for top-level documentation on this
+ function.
+
+ """
querycontext = util.preloaded.orm_context
return result
+ @util.deprecated_20(
+ ":meth:`_orm.Query.merge_result`",
+ alternative="The method is superseded by the "
+ ":func:`_orm.merge_frozen_result` function.",
+ becomes_legacy=True,
+ enable_warnings=False, # warnings occur via loading.merge_result
+ )
def merge_result(self, iterator, load=True):
"""Merge a result into this :class:`_query.Query` object's Session.
message=r"^The (Sybase|firebird) dialect is deprecated and will be",
)
- # 2.0 deprecation warnings, which we will want to have all of these
- # be "error" however for I98b8defdf7c37b818b3824d02f7668e3f5f31c94
- # we are moving one at a time
- for msg in [
- #
- # ORM Session
- #
- r"The merge_result\(\) method is superseded by the "
- r"merge_frozen_result\(\)",
- ]:
- warnings.filterwarnings(
- "ignore",
- message=msg,
- category=sa_exc.RemovedIn20Warning,
- )
-
try:
import pytest
except ImportError:
def _warn_with_version(msg, version, type_, stacklevel, code=None):
- if issubclass(type_, exc.RemovedIn20Warning) and not SQLALCHEMY_WARN_20:
+ if (
+ issubclass(type_, exc.Base20DeprecationWarning)
+ and not SQLALCHEMY_WARN_20
+ ):
return
warn = type_(msg, code=code)
if alternative:
message += " " + alternative
- return deprecated(
- "2.0", message=message, warning=exc.RemovedIn20Warning, **kw
- )
+ if becomes_legacy:
+ warning_cls = exc.LegacyAPIWarning
+ else:
+ warning_cls = exc.RemovedIn20Warning
+
+ return deprecated("2.0", message=message, warning=warning_cls, **kw)
def deprecated_params(**specs):
if constructor is not None:
docstring_header %= dict(func=constructor)
- if issubclass(wtype, exc.RemovedIn20Warning):
+ if issubclass(wtype, exc.Base20DeprecationWarning):
docstring_header += (
" (Background on SQLAlchemy 2.0 at: "
":ref:`migration_20_toplevel`)"
message = _sanitize_restructured_text(message)
- if issubclass(wtype, exc.RemovedIn20Warning):
+ if issubclass(wtype, exc.Base20DeprecationWarning):
doc_only = (
" (Background on SQLAlchemy 2.0 at: "
":ref:`migration_20_toplevel`)"
from sqlalchemy import true
from sqlalchemy import util
from sqlalchemy.engine import default
+from sqlalchemy.engine import result_tuple
from sqlalchemy.engine.base import Engine
from sqlalchemy.orm import aliased
from sqlalchemy.orm import as_declarative
r"the 1.x series of SQLAlchemy and will be removed in 2.0"
)
+merge_result_dep = (
+ r"The merge_result\(\) function is considered legacy as of the 1.x series"
+)
+
def _aliased_join_warning(arg=None):
return testing.expect_warnings(
assert (
c.exec_driver_sql("select count(1) from users").scalar() == 1
)
+
+
+class MergeResultTest(_fixtures.FixtureTest):
+ run_setup_mappers = "once"
+ run_inserts = "once"
+ run_deletes = None
+
+ @classmethod
+ def setup_mappers(cls):
+ cls._setup_stock_mapping()
+
+ def _fixture(self):
+ User = self.classes.User
+
+ s = fixture_session()
+ u1, u2, u3, u4 = (
+ User(id=1, name="u1"),
+ User(id=2, name="u2"),
+ User(id=7, name="u3"),
+ User(id=8, name="u4"),
+ )
+ s.query(User).filter(User.id.in_([7, 8])).all()
+ s.close()
+ return s, [u1, u2, u3, u4]
+
+ def test_single_entity(self):
+ s, (u1, u2, u3, u4) = self._fixture()
+ User = self.classes.User
+
+ q = s.query(User)
+ collection = [u1, u2, u3, u4]
+
+ with assertions.expect_deprecated_20(merge_result_dep):
+ it = q.merge_result(collection)
+ eq_([x.id for x in it], [1, 2, 7, 8])
+
+ def test_single_column(self):
+ User = self.classes.User
+
+ s = fixture_session()
+
+ q = s.query(User.id)
+ collection = [(1,), (2,), (7,), (8,)]
+ with assertions.expect_deprecated_20(merge_result_dep):
+ it = q.merge_result(collection)
+ eq_(list(it), [(1,), (2,), (7,), (8,)])
+
+ def test_entity_col_mix_plain_tuple(self):
+ s, (u1, u2, u3, u4) = self._fixture()
+ User = self.classes.User
+
+ q = s.query(User, User.id)
+ collection = [(u1, 1), (u2, 2), (u3, 7), (u4, 8)]
+ with assertions.expect_deprecated_20(merge_result_dep):
+ it = q.merge_result(collection)
+ it = list(it)
+ eq_([(x.id, y) for x, y in it], [(1, 1), (2, 2), (7, 7), (8, 8)])
+ eq_(list(it[0]._mapping.keys()), ["User", "id"])
+
+ def test_entity_col_mix_keyed_tuple(self):
+ s, (u1, u2, u3, u4) = self._fixture()
+ User = self.classes.User
+
+ q = s.query(User, User.id)
+
+ row = result_tuple(["User", "id"])
+
+ def kt(*x):
+ return row(x)
+
+ collection = [kt(u1, 1), kt(u2, 2), kt(u3, 7), kt(u4, 8)]
+ with assertions.expect_deprecated_20(merge_result_dep):
+ it = q.merge_result(collection)
+ it = list(it)
+ eq_([(x.id, y) for x, y in it], [(1, 1), (2, 2), (7, 7), (8, 8)])
+ eq_(list(it[0]._mapping.keys()), ["User", "id"])
+
+ def test_none_entity(self):
+ s, (u1, u2, u3, u4) = self._fixture()
+ User = self.classes.User
+
+ ua = aliased(User)
+ q = s.query(User, ua)
+
+ row = result_tuple(["User", "useralias"])
+
+ def kt(*x):
+ return row(x)
+
+ collection = [kt(u1, u2), kt(u1, None), kt(u2, u3)]
+ with assertions.expect_deprecated_20(merge_result_dep):
+ it = q.merge_result(collection)
+ eq_(
+ [(x and x.id or None, y and y.id or None) for x, y in it],
+ [(u1.id, u2.id), (u1.id, None), (u2.id, u3.id)],
+ )
from sqlalchemy import exc
from sqlalchemy import select
from sqlalchemy import testing
-from sqlalchemy.engine import result_tuple
-from sqlalchemy.orm import aliased
from sqlalchemy.orm import loading
from sqlalchemy.orm import relationship
from sqlalchemy.testing import mock
s.close()
return s, [u1, u2, u3, u4]
- def test_single_entity(self):
- s, (u1, u2, u3, u4) = self._fixture()
- User = self.classes.User
-
- q = s.query(User)
- collection = [u1, u2, u3, u4]
- it = loading.merge_result(q, collection)
- eq_([x.id for x in it], [1, 2, 7, 8])
-
def test_single_entity_frozen(self):
s = fixture_session()
User = self.classes.User
it = loading.merge_frozen_result(s, stmt, result.freeze())
eq_([x.id for x in it().scalars()], [7, 8, 9])
- def test_single_column(self):
- User = self.classes.User
-
- s = fixture_session()
-
- q = s.query(User.id)
- collection = [(1,), (2,), (7,), (8,)]
- it = loading.merge_result(q, collection)
- eq_(list(it), [(1,), (2,), (7,), (8,)])
-
def test_single_column_frozen(self):
User = self.classes.User
it = loading.merge_frozen_result(s, stmt, result.freeze())
eq_([x.id for x in it()], [7, 8, 9])
- def test_entity_col_mix_plain_tuple(self):
- s, (u1, u2, u3, u4) = self._fixture()
- User = self.classes.User
-
- q = s.query(User, User.id)
- collection = [(u1, 1), (u2, 2), (u3, 7), (u4, 8)]
- it = loading.merge_result(q, collection)
- it = list(it)
- eq_([(x.id, y) for x, y in it], [(1, 1), (2, 2), (7, 7), (8, 8)])
- eq_(list(it[0]._mapping.keys()), ["User", "id"])
-
def test_entity_col_mix_plain_tuple_frozen(self):
s = fixture_session()
User = self.classes.User
it = list(it())
eq_([(x.id, y) for x, y in it], [(7, 7), (8, 8), (9, 9)])
eq_(list(it[0]._mapping.keys()), ["User", "id"])
-
- def test_entity_col_mix_keyed_tuple(self):
- s, (u1, u2, u3, u4) = self._fixture()
- User = self.classes.User
-
- q = s.query(User, User.id)
-
- row = result_tuple(["User", "id"])
-
- def kt(*x):
- return row(x)
-
- collection = [kt(u1, 1), kt(u2, 2), kt(u3, 7), kt(u4, 8)]
- it = loading.merge_result(q, collection)
- it = list(it)
- eq_([(x.id, y) for x, y in it], [(1, 1), (2, 2), (7, 7), (8, 8)])
- eq_(list(it[0]._mapping.keys()), ["User", "id"])
-
- def test_none_entity(self):
- s, (u1, u2, u3, u4) = self._fixture()
- User = self.classes.User
-
- ua = aliased(User)
- q = s.query(User, ua)
-
- row = result_tuple(["User", "useralias"])
-
- def kt(*x):
- return row(x)
-
- collection = [kt(u1, u2), kt(u1, None), kt(u2, u3)]
- it = loading.merge_result(q, collection)
- eq_(
- [(x and x.id or None, y and y.id or None) for x, y in it],
- [(u1.id, u2.id), (u1.id, None), (u2.id, u3.id)],
- )
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
