added validation to delete queries to make sure they have a where clause

Author: dantownsend

docs/src/piccolo/query_types/delete.rst

@@ -10,6 +10,22 @@ This deletes any matching rows from the table.
     >>> Band.delete().where(Band.name == 'Rustaceans').run_sync()
     []
 
+force
+-----
+
+Piccolo won't let you run a delete query without a where clause, unless you
+explicitly tell it to do so. This is to help prevent accidentally deleting all
+the data from a table.
+
+.. code-block:: python
+
+    >>> Band.delete().run_sync()
+    Raises: DeletionError
+
+    # Works fine:
+    >>> Band.delete(force=True).run_sync()
+    []
+
 Query clauses
 -------------
 

piccolo/query/base.py

@@ -74,7 +74,17 @@ def _process_results(self, results):
 
         return raw
 
+    def _validate(self):
+        """
+        Override in any subclasses if validation needs to be run before
+        executing a query - for example, warning a user if they're about to
+        delete all the data from a table.
+        """
+        pass
+
     async def run(self, in_pool=True):
+        self._validate()
+
         engine = getattr(self.table._meta, "db", None)
         if not engine:
             raise ValueError(

piccolo/query/methods/delete.py

@@ -6,9 +6,17 @@
 from piccolo.querystring import QueryString
 
 
+class DeletionError(Exception):
+    pass
+
+
 class Delete(Query):
 
-    __slots__ = ("where_delegate",)
+    __slots__ = ("where_delegate", "force")
+
+    def __init__(self, force=False, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.force = force
 
     def setup_delegates(self):
         self.where_delegate = WhereDelegate()
@@ -17,6 +25,18 @@ def where(self, where: Combinable) -> Delete:
         self.where_delegate.where(where)
         return self
 
+    def _validate(self):
+        """
+        Don't let a deletion happen unless it has a where clause, or is
+        explicitly forced.
+        """
+        if (not self.where_delegate._where) and (not self.force):
+            raise DeletionError(
+                "Warning - do you really want to delete all the data from "
+                f"{self.table._meta.tablename}? If so, use "
+                "MyTable.delete(force=True), or add a where clause."
+            )
+
     @property
     def querystring(self) -> QueryString:
         query = f"DELETE FROM {self.table._meta.tablename}"

piccolo/table.py

@@ -295,11 +295,16 @@ def select(cls) -> Select:
         return Select(table=cls)
 
     @classmethod
-    def delete(cls) -> Delete:
+    def delete(cls, force=False) -> Delete:
         """
+        Delete rows from the table.
+
         await Band.delete().where(Band.name == 'Pythonistas').run()
+
+        Unless 'force' is set to True, deletions aren't allowed without a
+        'where' clause, to prevent accidental mass deletions.
         """
-        return Delete(table=cls)
+        return Delete(table=cls, force=force)
 
     @classmethod
     def create(cls) -> Create:

tests/table/test_delete.py

@@ -1,3 +1,5 @@
+from piccolo.query.methods.delete import DeletionError
+
 from ..base import DBTestCase
 from ..example_project.tables import Band
 
@@ -12,3 +14,12 @@ def test_delete(self):
         print(f"response = {response}")
 
         self.assertEqual(response, 0)
+
+    def test_validation(self):
+        """
+        Make sure you can't delete all the data without forcing it.
+        """
+        with self.assertRaises(DeletionError):
+            Band.delete().run_sync()
+
+        Band.delete(force=True).run_sync()