storage endpoint for deleting multiple runs in one query

python_modules/dagster/dagster/_core/storage/legacy_storage.py

@@ -288,6 +288,9 @@ def wipe(self) -> None:
     def delete_run(self, run_id: str) -> None:
         return self._storage.run_storage.delete_run(run_id)
 
+    def delete_runs(self, run_ids: Sequence[str]) -> None:
+        return self._storage.run_storage.delete_runs(run_ids)
+
     def migrate(self, print_fn: Optional[PrintFn] = None, force_rebuild_all: bool = False) -> None:
         return self._storage.run_storage.migrate(print_fn, force_rebuild_all)
 

python_modules/dagster/dagster/_core/storage/runs/base.py

@@ -318,6 +318,10 @@ def wipe(self) -> None:
     def delete_run(self, run_id: str) -> None:
         """Remove a run from storage."""
 
+    @abstractmethod
+    def delete_runs(self, run_ids: Sequence[str]) -> None:
+        """Remove a list of runs from storage."""
+
     @property
     def supports_bucket_queries(self) -> bool:
         return False

python_modules/dagster/dagster/_core/storage/runs/sql_run_storage.py

@@ -562,6 +562,12 @@ def delete_run(self, run_id: str) -> None:
         with self.connect() as conn:
             conn.execute(query)
 
+    def delete_runs(self, run_ids: Sequence[str]) -> None:
+        check.list_param(run_ids, "run_ids", of_type=str)
+        query = db.delete(RunsTable).where(RunsTable.c.run_id.in_(run_ids))
+        with self.connect() as conn:
+            conn.execute(query)
+
     def has_job_snapshot(self, job_snapshot_id: str) -> bool:
         check.str_param(job_snapshot_id, "job_snapshot_id")
         return self._has_snapshot_id(job_snapshot_id)

python_modules/dagster/dagster_tests/storage_tests/utils/run_storage.py

@@ -1032,6 +1032,24 @@ def test_delete(self, storage):
         storage.delete_run(run_id)
         assert list(storage.get_runs()) == []
 
+    def test_delete_multiple(self, storage):
+        if not self.can_delete_runs():
+            pytest.skip("storage cannot delete runs")
+
+        assert storage
+        run_ids = []
+        for _ in range(3):
+            run_id = make_new_run_id()
+            storage.add_run(TestRunStorage.build_run(run_id=run_id, job_name="some_pipeline"))
+            run_ids.append(run_id)
+
+        storage.add_run(
+            TestRunStorage.build_run(run_id=make_new_run_id(), job_name="some_pipeline")
+        )
+        assert len(storage.get_runs()) == 4
+        storage.delete_runs(run_ids)
+        assert len(storage.get_runs()) == 1
+
     def test_delete_with_tags(self, storage: RunStorage):
         if not self.can_delete_runs():
             pytest.skip("storage cannot delete runs")
@@ -1051,6 +1069,34 @@ def test_delete_with_tags(self, storage: RunStorage):
         assert list(storage.get_runs()) == []
         assert run_id not in [key for key, value in storage.get_run_tags(tag_keys=[run_id])]
 
+    def test_delete_multiple_with_tags(self, storage: RunStorage):
+        if not self.can_delete_runs():
+            pytest.skip("storage cannot delete runs")
+
+        assert storage
+        run_ids = []
+        for _ in range(3):
+            run_id = make_new_run_id()
+            storage.add_run(
+                TestRunStorage.build_run(
+                    run_id=run_id,
+                    job_name="some_pipeline",
+                    tags={run_id: run_id},
+                )
+            )
+            run_ids.append(run_id)
+
+        storage.add_run(
+            TestRunStorage.build_run(
+                run_id=make_new_run_id(),
+                job_name="some_pipeline",
+                tags={"not_deleted": "true"},
+            )
+        )
+        assert len(storage.get_runs()) == 4
+        storage.delete_runs(run_ids)
+        assert len(storage.get_runs()) == 1
+
     def test_wipe_tags(self, storage: RunStorage):
         if not self.can_delete_runs():
             pytest.skip("storage cannot delete")