code: def test_dag_run_custom_sqla_interface_delete_no_collateral_damage(
      dag_maker, session
  ):
      interface = DagRunCustomSQLAInterface(obj=DagRun, session=session)
      dag_ids = (f"test_dag_{x}" for x in range(1, 4))
      dates = (pendulum.datetime(2023, 1, x) for x in range(1, 4))
      for dag_id, date in itertools.product(dag_ids, dates):
          with dag_maker(dag_id=dag_id) as dag:
              dag.create_dagrun(
                  execution_date=date,
                  state="running",
                  run_type="scheduled",
              )
      dag_runs = session.query(DagRun).all()
      assert len(dag_runs) == 9
      assert len(set(x.run_id for x in dag_runs)) == 3
      run_id_for_single_delete = "scheduled__2023-01-01T00:00:00+00:00"
      # we have 3 runs with this same run_id
      assert (
          sum(
              1
              for x in dag_runs
              if x.run_id == run_id_for_single_delete
          )
          == 3
      )
      # each is a different dag
  
      # if we delete one, it shouldn't delete the others
      one_run = next(
          x for x in dag_runs if x.run_id == run_id_for_single_delete
      )
      assert interface.delete(item=one_run) is True
      session.commit()
      dag_runs = session.query(DagRun).all()
      # we should have one fewer dag run now
      assert len(dag_runs) == 8
  
      # now let's try multi delete
      run_id_for_multi_delete = "scheduled__2023-01-02T00:00:00+00:00"
      # verify we have 3
      runs_of_interest = [
          x for x in dag_runs if x.run_id == run_id_for_multi_delete
      ]
      assert len(runs_of_interest) == 3
      # and that each is different dag
      assert len(set(x.dag_id for x in dag_runs)) == 3
  
      to_delete = runs_of_interest[:2]
      # now try multi delete
      assert interface.delete_all(items=to_delete) is True
      session.commit()
      dag_runs = session.query(DagRun).all()
      assert len(dag_runs) == 6
      assert len(set(x.dag_id for x in dag_runs)) == 3
      assert len(set(x.run_id for x in dag_runs)) == 3
  
imports: from __future__ import annotations
  import itertools
  import re
  from datetime import datetime
  from unittest.mock import Mock
  from urllib.parse import parse_qs
  import pendulum
  import pytest
  from bs4 import BeautifulSoup
  from markupsafe import Markup
  from airflow.models import DagRun
  from airflow.utils import json
  from airflow.www import utils
  from airflow.www.utils import DagRunCustomSQLAInterface, json_f, wrapped_markdown
  from tests.test_utils.config import conf_vars
label_desc: Use image node_id 1 for calling a global function with example usage: test_dag_run_custom_sqla_interface_delete_no_collateral_damage(dag_maker, session) without return types