"""
self._execution_options = self._execution_options.union(kwargs)
- @_generative
- @util.deprecated(
- "0.9",
- "The :meth:`.Query.with_lockmode` method is deprecated and will "
- "be removed in a future release. Please refer to "
- ":meth:`.Query.with_for_update`. ",
- )
- def with_lockmode(self, mode):
- """Return a new :class:`.Query` object with the specified "locking mode",
- which essentially refers to the ``FOR UPDATE`` clause.
-
- :param mode: a string representing the desired locking mode.
- Valid values are:
-
- * ``None`` - translates to no lockmode
-
- * ``'update'`` - translates to ``FOR UPDATE``
- (standard SQL, supported by most dialects)
-
- * ``'update_nowait'`` - translates to ``FOR UPDATE NOWAIT``
- (supported by Oracle, PostgreSQL 8.1 upwards)
-
- * ``'read'`` - translates to ``LOCK IN SHARE MODE`` (for MySQL),
- and ``FOR SHARE`` (for PostgreSQL)
-
- .. seealso::
-
- :meth:`.Query.with_for_update` - improved API for
- specifying the ``FOR UPDATE`` clause.
-
- """
- self._for_update_arg = LockmodeArg.parse_legacy_query(mode)
-
@_generative
def with_for_update(
self,
SELECT users.id AS users_id FROM users FOR UPDATE OF users NOWAIT
- .. versionadded:: 0.9.0 :meth:`.Query.with_for_update` supersedes
- the :meth:`.Query.with_lockmode` method.
-
.. seealso::
:meth:`.GenerativeSelect.with_for_update` - Core level method with
full argument and behavioral description.
"""
- self._for_update_arg = LockmodeArg(
+ self._for_update_arg = ForUpdateArg(
read=read,
nowait=nowait,
of=of,
)
-class LockmodeArg(ForUpdateArg):
- @classmethod
- def parse_legacy_query(cls, mode):
- if mode in (None, False):
- return None
-
- if mode == "read":
- read = True
- nowait = False
- elif mode == "update":
- read = nowait = False
- elif mode == "update_nowait":
- nowait = True
- read = False
- else:
- raise sa_exc.ArgumentError(
- "Unknown with_lockmode argument: %r" % mode
- )
-
- return LockmodeArg(read=read, nowait=nowait)
-
-
class _QueryEntity(object):
"""represent an entity column returned within a Query result."""
util.raise_(e, with_traceback=sys.exc_info()[2])
def refresh(
- self,
- instance,
- attribute_names=None,
- with_for_update=None,
- lockmode=None,
+ self, instance, attribute_names=None, with_for_update=None,
):
"""Expire and refresh the attributes on the given instance.
.. versionadded:: 1.2
- :param lockmode: Passed to the :class:`~sqlalchemy.orm.query.Query`
- as used by :meth:`~sqlalchemy.orm.query.Query.with_lockmode`.
- Superseded by :paramref:`.Session.refresh.with_for_update`.
-
.. seealso::
:ref:`session_expire` - introductory material
"A blank dictionary is ambiguous."
)
- if lockmode:
- with_for_update = query.LockmodeArg.parse_legacy_query(lockmode)
- elif with_for_update is not None:
+ if with_for_update is not None:
if with_for_update is True:
- with_for_update = query.LockmodeArg()
+ with_for_update = query.ForUpdateArg()
elif with_for_update:
- with_for_update = query.LockmodeArg(**with_for_update)
+ with_for_update = query.ForUpdateArg(**with_for_update)
else:
with_for_update = None
),
)
@_document_text_coercion("text", ":func:`.text`", ":paramref:`.text.text`")
- def _create_text(
- self, text, bind=None, bindparams=None, typemap=None,
- ):
+ def _create_text(self, text, bind=None, bindparams=None, typemap=None):
r"""Construct a new :class:`.TextClause` clause, representing
a textual SQL string directly.
return self._join_condition(left, right, a_subset=left_right)
@classmethod
- @util.deprecated_params(
- ignore_nonexistent_tables=(
- "0.9",
- "The :paramref:`.join_condition.ignore_nonexistent_tables` "
- "parameter is deprecated and will be removed in a future "
- "release. Tables outside of the two tables being handled "
- "are no longer considered.",
- )
- )
def _join_condition(
- cls,
- a,
- b,
- ignore_nonexistent_tables=False,
- a_subset=None,
- consider_as_foreign_keys=None,
+ cls, a, b, a_subset=None, consider_as_foreign_keys=None
):
"""create a join condition between two tables or selectables.
between the two selectables. If there are multiple ways
to join, or no way to join, an error is raised.
- :param ignore_nonexistent_tables: unused - tables outside of the
- two tables being handled are not considered.
-
:param a_subset: An optional expression that is a sub-component
of ``a``. An attempt will be made to join to just this sub-component
first before looking at the full ``a`` construct, and if found
argument as a no-op, so that the argument can be passed to the
``alias()`` method of any selectable.
- .. versionadded:: 0.9.0 Added the ``flat=True`` option to create
- "aliases" of joins without enclosing inside of a SELECT
- subquery.
-
:param name: name given to the alias.
:param flat: if True, produce an alias of the left and right
two selectables. This produces join expression that does not
include an enclosing SELECT.
- .. versionadded:: 0.9.0
-
.. seealso::
:ref:`core_tutorial_aliases`
is an instance of :class:`.Join` - see :meth:`.Join.alias`
for details.
- .. versionadded:: 0.9.0
-
"""
return coercions.expect(
roles.FromClauseRole, selectable, allow_select=True
):
"""Represents arguments specified to :meth:`.Select.for_update`.
- .. versionadded:: 0.9.0
-
"""
self.nowait = nowait
represents a fixed textual string which cannot be altered at this level,
only wrapped as a subquery.
- .. versionadded:: 0.9.0 :class:`.GenerativeSelect` was added to
- provide functionality specific to :class:`.Select` and
- :class:`.CompoundSelect` while allowing :class:`.SelectBase` to be
- used for other SELECT-like objects, e.g. :class:`.TextualSelect`.
-
"""
_order_by_clauses = ()
import sqlalchemy as sa
from sqlalchemy import and_
from sqlalchemy import event
-from sqlalchemy import exc
from sqlalchemy import func
from sqlalchemy import Integer
from sqlalchemy import select
# same here, this was "passing string names to Query.columns"
# deprecation message, that's gone here?
assert_raises_message(
- exc.ArgumentError,
+ sa.exc.ArgumentError,
"Textual column expression 'name' should be explicitly",
s.query,
User.id,
sess.query(User).options(undefer("addresses", "email_address"))
-class LegacyLockModeTest(_fixtures.FixtureTest):
- run_inserts = None
-
- @classmethod
- def setup_mappers(cls):
- User, users = cls.classes.User, cls.tables.users
- mapper(User, users)
-
- def _assert_legacy(self, arg, read=False, nowait=False):
- User = self.classes.User
- s = Session()
-
- with testing.expect_deprecated(
- r"The Query.with_lockmode\(\) method is deprecated"
- ):
- q = s.query(User).with_lockmode(arg)
- sel = q._compile_context().statement
-
- if arg is None:
- assert q._for_update_arg is None
- assert sel._for_update_arg is None
- return
-
- assert q._for_update_arg.read is read
- assert q._for_update_arg.nowait is nowait
-
- assert sel._for_update_arg.read is read
- assert sel._for_update_arg.nowait is nowait
-
- def test_false_legacy(self):
- self._assert_legacy(None)
-
- def test_plain_legacy(self):
- self._assert_legacy("update")
-
- def test_nowait_legacy(self):
- self._assert_legacy("update_nowait", nowait=True)
-
- def test_read_legacy(self):
- self._assert_legacy("read", read=True)
-
- def test_unknown_legacy_lock_mode(self):
- User = self.classes.User
- sess = Session()
- with testing.expect_deprecated(
- r"The Query.with_lockmode\(\) method is deprecated"
- ):
- assert_raises_message(
- exc.ArgumentError,
- "Unknown with_lockmode argument: 'unknown_mode'",
- sess.query(User.id).with_lockmode,
- "unknown_mode",
- )
-
-
class InstrumentationTest(fixtures.ORMTest):
def test_dict_subclass4(self):
# tests #2654
s.refresh(m1, with_for_update=False)
s.refresh(m1)
- from sqlalchemy.orm.query import LockmodeArg
+ from sqlalchemy.orm.query import ForUpdateArg
eq_(
[
call[-1]["with_for_update"]
for call in load_on_ident.mock_calls
],
- [LockmodeArg(read=True), LockmodeArg(), None, None],
+ [ForUpdateArg(read=True), ForUpdateArg(), None, None],
)
s1.close()
s1.query(Foo).with_for_update(read=True).get(f1s1.id)
- @engines.close_open_connections
- def test_versioncheck_legacy(self):
- """query.with_lockmode performs a 'version check' on an already loaded
- instance"""
-
- Foo = self.classes.Foo
-
- s1 = self._fixture()
- f1s1 = Foo(value="f1 value")
- s1.add(f1s1)
- s1.commit()
-
- s2 = create_session(autocommit=False)
- f1s2 = s2.query(Foo).get(f1s1.id)
- f1s2.value = "f1 new value"
- with conditional_sane_rowcount_warnings(
- update=True, only_returning=True
- ):
- s2.commit()
-
- # load, version is wrong
- assert_raises_message(
- sa.orm.exc.StaleDataError,
- r"Instance .* has version id '\d+' which does not "
- r"match database-loaded version id '\d+'",
- s1.query(Foo).with_for_update(read=True).get,
- f1s1.id,
- )
-
- # reload it - this expires the old version first
- s1.refresh(f1s1, with_for_update=dict(read=True))
-
- # now assert version OK
- s1.query(Foo).with_for_update(read=True).get(f1s1.id)
-
- # assert brand new load is OK too
- s1.close()
- s1.query(Foo).with_for_update(read=True).get(f1s1.id)
-
def test_versioncheck_not_versioned(self):
"""ensure the versioncheck logic skips if there isn't a
version_id_col actually configured"""
f1s2.value = "f1 new value"
assert_raises(
- exc.DBAPIError, s1.refresh, f1s1, lockmode="update_nowait"
+ exc.DBAPIError, s1.refresh, f1s1, with_for_update={"nowait": True}
)
s1.rollback()
s1.refresh(f1s1, with_for_update={"nowait": True})
assert f1s1.version_id == f1s2.version_id
- @engines.close_open_connections
- @testing.requires.update_nowait
- def test_versioncheck_for_update_legacy(self):
- """query.with_lockmode performs a 'version check' on an already loaded
- instance"""
-
- Foo = self.classes.Foo
-
- s1 = self._fixture()
- f1s1 = Foo(value="f1 value")
- s1.add(f1s1)
- s1.commit()
-
- s2 = create_session(autocommit=False)
- f1s2 = s2.query(Foo).get(f1s1.id)
- s2.refresh(f1s2, lockmode="update")
- f1s2.value = "f1 new value"
-
- assert_raises(
- exc.DBAPIError, s1.refresh, f1s1, lockmode="update_nowait"
- )
- s1.rollback()
-
- with conditional_sane_rowcount_warnings(update=True):
- s2.commit()
- s1.refresh(f1s1, lockmode="update_nowait")
- assert f1s1.version_id == f1s2.version_id
-
def test_update_multi_missing_broken_multi_rowcount(self):
@util.memoized_property
def rowcount(self):
from sqlalchemy.sql import coercions
from sqlalchemy.sql import quoted_name
from sqlalchemy.sql import roles
-from sqlalchemy.sql import util as sql_util
from sqlalchemy.sql import visitors
from sqlalchemy.sql.selectable import SelectStatementGrouping
from sqlalchemy.testing import assert_raises
):
create_engine("mysql://", convert_unicode=True, module=mock.Mock())
- def test_join_condition_ignore_nonexistent_tables(self):
- m = MetaData()
- t1 = Table("t1", m, Column("id", Integer))
- t2 = Table(
- "t2", m, Column("id", Integer), Column("t1id", ForeignKey("t1.id"))
- )
- with testing.expect_deprecated(
- "The join_condition.ignore_nonexistent_tables "
- "parameter is deprecated"
- ):
- join_cond = sql_util.join_condition(
- t1, t2, ignore_nonexistent_tables=True
- )
-
- t1t2 = t1.join(t2)
-
- assert t1t2.onclause.compare(join_cond)
-
def test_empty_and_or(self):
with testing.expect_deprecated(
r"Invoking and_\(\) without arguments is deprecated, and "
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
