.. changelog::
:version: 0.9.10
+ .. change::
+ :tags: bug, orm
+ :tickets: 3349
+
+ :class:`.Query` doesn't support joins, subselects, or special
+ FROM clauses when using the :meth:`.Query.update` or
+ :meth:`.Query.delete` methods; instead of silently ignoring these
+ fields if methods like :meth:`.Query.join` or
+ :meth:`.Query.select_from` has been called, a warning is emitted.
+ As of 1.0.0b5 this will raise an error.
+
.. change::
:tags: bug, mysql, pymysql
:tickets: 3337
from . import attributes, sync, exc as orm_exc, evaluator
from .base import _state_mapper, state_str, _attr_as_key
from ..sql import expression
+from ..sql.base import _from_objects
from . import loading
def __init__(self, query):
self.query = query.enable_eagerloads(False)
+ self._validate_query_state()
+
+ def _validate_query_state(self):
+ for attr, methname, notset, err in (
+ ('_limit', 'limit()', None, True),
+ ('_offset', 'offset()', None, True),
+ ('_order_by', 'order_by()', False, True),
+ ('_group_by', 'group_by()', False, True),
+ ('_distinct', 'distinct()', False, True),
+ (
+ '_from_obj',
+ 'join(), outerjoin(), select_from(), or from_self()',
+ (), False)
+ ):
+ if getattr(self.query, attr) is not notset:
+ if err:
+ raise sa_exc.InvalidRequestError(
+ "Can't call Query.update() or Query.delete() "
+ "when %s has been called" %
+ (methname, )
+ )
+ else:
+ util.warn(
+ "Can't call Query.update() or Query.delete() "
+ "when %s has been called. This will be an "
+ "exception in 1.0" %
+ (methname, )
+ )
@property
def session(self):
def _do_pre(self):
query = self.query
+
+ # the completion of the full Query statement here is wasteful;
+ # in 1.0 we will directly extract the target table without
+ # this step.
self.context = context = query._compile_context()
if len(context.statement.froms) != 1 or \
not isinstance(context.statement.froms[0], schema.Table):
def __init__(self, query, values):
super(BulkUpdate, self).__init__(query)
- self.query._no_select_modifiers("update")
self.values = values
@classmethod
def __init__(self, query):
super(BulkDelete, self).__init__(query)
- self.query._no_select_modifiers("delete")
@classmethod
def factory(cls, query, synchronize_session):
% (meth, meth)
)
- def _no_select_modifiers(self, meth):
- if not self._enable_assertions:
- return
- for attr, methname, notset in (
- ('_limit', 'limit()', None),
- ('_offset', 'offset()', None),
- ('_order_by', 'order_by()', False),
- ('_group_by', 'group_by()', False),
- ('_distinct', 'distinct()', False),
- ):
- if getattr(self, attr) is not notset:
- raise sa_exc.InvalidRequestError(
- "Can't call Query.%s() when %s has been called" %
- (meth, methname)
- )
-
def _get_options(self, populate_existing=None,
version_check=None,
only_load_props=None,
text("select * from table"))
assert_raises(sa_exc.InvalidRequestError, q.with_polymorphic, User)
- def test_cancel_order_by(self):
- User = self.classes.User
-
- s = create_session()
-
- q = s.query(User).order_by(User.id)
- self.assert_compile(
- q,
- "SELECT users.id AS users_id, users.name AS users_name "
- "FROM users ORDER BY users.id",
- use_default_dialect=True)
-
- assert_raises(
- sa_exc.InvalidRequestError, q._no_select_modifiers, "foo")
-
- q = q.order_by(None)
- self.assert_compile(
- q,
- "SELECT users.id AS users_id, users.name AS users_name FROM users",
- use_default_dialect=True)
-
- assert_raises(
- sa_exc.InvalidRequestError, q._no_select_modifiers, "foo")
-
- q = q.order_by(False)
- self.assert_compile(
- q,
- "SELECT users.id AS users_id, users.name AS users_name FROM users",
- use_default_dialect=True)
-
- # after False was set, this should pass
- q._no_select_modifiers("foo")
-
def test_mapper_zero(self):
User, Address = self.classes.User, self.classes.Address
test_needs_autoincrement=True),
Column('name', String(32)),
Column('age', Integer))
+ Table(
+ "addresses", metadata,
+ Column('id', Integer, primary_key=True),
+ Column('user_id', ForeignKey('users.id'))
+ )
@classmethod
def setup_classes(cls):
class User(cls.Comparable):
pass
+ class Address(cls.Comparable):
+ pass
+
@classmethod
def insert_data(cls):
users = cls.tables.users
User = cls.classes.User
users = cls.tables.users
- mapper(User, users)
+ Address = cls.classes.Address
+ addresses = cls.tables.addresses
+
+ mapper(User, users, properties={
+ 'addresses': relationship(Address)
+ })
+ mapper(Address, addresses)
def test_illegal_eval(self):
User = self.classes.User
def test_illegal_operations(self):
User = self.classes.User
+ Address = self.classes.Address
s = Session()
- for q, mname in (
- (s.query(User).limit(2), "limit"),
- (s.query(User).offset(2), "offset"),
- (s.query(User).limit(2).offset(2), "limit"),
- (s.query(User).order_by(User.id), "order_by"),
- (s.query(User).group_by(User.id), "group_by"),
- (s.query(User).distinct(), "distinct")
+ for q, mname, err in (
+ (s.query(User).limit(2), r"limit\(\)", True),
+ (s.query(User).offset(2), r"offset\(\)", True),
+ (s.query(User).limit(2).offset(2), r"limit\(\)", True),
+ (s.query(User).order_by(User.id), r"order_by\(\)", True),
+ (s.query(User).group_by(User.id), r"group_by\(\)", True),
+ (s.query(User).distinct(), r"distinct\(\)", True),
+ (s.query(User).join(User.addresses),
+ r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
+ False),
+ (s.query(User).outerjoin(User.addresses),
+ r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
+ False),
+ (s.query(User).select_from(Address),
+ r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
+ False),
+ (s.query(User).from_self(),
+ r"join\(\), outerjoin\(\), select_from\(\), or from_self\(\)",
+ False),
):
+ if err:
+ exc_cls = exc.InvalidRequestError
+ else:
+ exc_cls = exc.SAWarning
+
assert_raises_message(
- exc.InvalidRequestError,
- r"Can't call Query.update\(\) when %s\(\) has been called" % mname,
+ exc_cls,
+ r"Can't call Query.update\(\) or Query.delete\(\) when "
+ "%s has been called" % mname,
q.update,
{'name': 'ed'})
assert_raises_message(
- exc.InvalidRequestError,
- r"Can't call Query.delete\(\) when %s\(\) has been called" % mname,
+ exc_cls,
+ r"Can't call Query.update\(\) or Query.delete\(\) when "
+ "%s has been called" % mname,
q.delete)
def test_delete(self):
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
