--- /dev/null
+.. change::
+ :tags: feature, orm
+ :tickets: 4461
+
+ Added new event hooks :meth:`.QueryEvents.before_compile_update` and
+ :meth:`.QueryEvents.before_compile_delete` which complement
+ :meth:`.QueryEvents.before_compile` in the case of the :meth:`.Query.update`
+ and :meth:`.Query.delete` methods.
+
The event should normally be listened with the ``retval=True``
parameter set, so that the modified query may be returned.
+ .. seealso::
+
+ :meth:`.QueryEvents.before_compile_update`
+
+ :meth:`.QueryEvents.before_compile_delete`
+
+
+ """
+
+ def before_compile_update(self, query, update_context):
+ """Allow modifications to the :class:`.Query` object within
+ :meth:`.Query.update`.
+
+ Like the :meth:`.QueryEvents.before_compile` event, this event
+ should be configured with ``retval=True``, and the modified
+ :class:`.Query` object returned, as in ::
+
+ @event.listens_for(Query, "before_compile_update", retval=True)
+ def no_deleted(query, update_context):
+ for desc in query.column_descriptions:
+ if desc['type'] is User:
+ entity = desc['entity']
+ query = query.filter(entity.deleted == False)
+ return query
+
+ :param query: a :class:`.Query` instance; this is also
+ the ``.query`` attribute of the given "update context"
+ object.
+
+ :param update_context: an "update context" object which is
+ the same kind of object as described in
+ :paramref:`.QueryEvents.after_bulk_update.update_context`.
+
+ .. versionadded:: 1.2.17
+
+ .. seealso::
+
+ :meth:`.QueryEvents.before_compile`
+
+ :meth:`.QueryEvents.before_compile_delete`
+
+
+ """
+
+ def before_compile_delete(self, query, delete_context):
+ """Allow modifications to the :class:`.Query` object within
+ :meth:`.Query.delete`.
+
+ Like the :meth:`.QueryEvents.before_compile` event, this event
+ should be configured with ``retval=True``, and the modified
+ :class:`.Query` object returned, as in ::
+
+ @event.listens_for(Query, "before_compile_delete", retval=True)
+ def no_deleted(query, delete_context):
+ for desc in query.column_descriptions:
+ if desc['type'] is User:
+ entity = desc['entity']
+ query = query.filter(entity.deleted == False)
+ return query
+
+ :param query: a :class:`.Query` instance; this is also
+ the ``.query`` attribute of the given "delete context"
+ object.
+
+ :param delete_context: a "delete context" object which is
+ the same kind of object as described in
+ :paramref:`.QueryEvents.after_bulk_delete.delete_context`.
+
+ .. versionadded:: 1.2.17
+
+ .. seealso::
+
+ :meth:`.QueryEvents.before_compile`
+
+ :meth:`.QueryEvents.before_compile_update`
+
"""
return klass(*arg)
def exec_(self):
+ self._do_before_compile()
self._do_pre()
self._do_pre_synchronize()
self._do_exec()
self.result = self.query._execute_crud(stmt, self.mapper)
self.rowcount = self.result.rowcount
+ def _do_before_compile(self):
+ raise NotImplementedError()
+
@util.dependencies("sqlalchemy.orm.query")
def _do_pre(self, querylib):
query = self.query
+
self.context = querylib.QueryContext(query)
if isinstance(query._entities[0], querylib._ColumnEntity):
update_kwargs,
)
+ def _do_before_compile(self):
+ if self.query.dispatch.before_compile_update:
+ for fn in self.query.dispatch.before_compile_update:
+ new_query = fn(self.query, self)
+ if new_query is not None:
+ self.query = new_query
+
@property
def _resolved_values(self):
values = []
query,
)
+ def _do_before_compile(self):
+ if self.query.dispatch.before_compile_delete:
+ for fn in self.query.dispatch.before_compile_delete:
+ new_query = fn(self.query, self)
+ if new_query is not None:
+ self.query = new_query
+
def _do_exec(self):
delete_stmt = sql.delete(self.primary_table, self.context.whereclause)
from sqlalchemy.testing import eq_
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_not_
+from sqlalchemy.testing.assertsql import CompiledSQL
from sqlalchemy.testing.mock import ANY
from sqlalchemy.testing.mock import call
from sqlalchemy.testing.mock import Mock
class QueryEventsTest(
- _RemoveListeners, _fixtures.FixtureTest, AssertsCompiledSQL
+ _RemoveListeners,
+ _fixtures.FixtureTest,
+ AssertsCompiledSQL,
+ testing.AssertsExecutionResults,
):
__dialect__ = "default"
)
eq_(q.all(), [(7, "jack")])
+ def test_before_compile_update(self):
+ @event.listens_for(query.Query, "before_compile_update", retval=True)
+ def no_deleted(query, update_context):
+ assert update_context.query is query
+
+ for desc in query.column_descriptions:
+ if desc["type"] is User:
+ entity = desc["expr"]
+ query = query.filter(entity.id != 10)
+ return query
+
+ User = self.classes.User
+ s = Session()
+
+ with self.sql_execution_asserter() as asserter:
+ q = s.query(User).filter_by(id=7).update({"name": "ed"})
+ asserter.assert_(
+ CompiledSQL(
+ "UPDATE users SET name=:name WHERE "
+ "users.id = :id_1 AND users.id != :id_2",
+ [{"name": "ed", "id_1": 7, "id_2": 10}],
+ )
+ )
+
+ def test_before_compile_delete(self):
+ @event.listens_for(query.Query, "before_compile_delete", retval=True)
+ def no_deleted(query, delete_context):
+ assert delete_context.query is query
+
+ for desc in query.column_descriptions:
+ if desc["type"] is User:
+ entity = desc["expr"]
+ query = query.filter(entity.id != 10)
+ return query
+
+ User = self.classes.User
+ s = Session()
+
+ # note this deletes no rows
+ with self.sql_execution_asserter() as asserter:
+ q = s.query(User).filter_by(id=10).delete()
+ asserter.assert_(
+ CompiledSQL(
+ "DELETE FROM users WHERE "
+ "users.id = :id_1 AND users.id != :id_2",
+ [{"id_1": 10, "id_2": 10}],
+ )
+ )
+
class RefreshFlushInReturningTest(fixtures.MappedTest):
"""test [ticket:3427].
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
