class_docstr: Base Trigger class.
  
  Triggers are a workload that run in an asynchronous event loop shared with
  other Triggers, and fire off events that will unpause deferred Tasks,
  start linked DAGs, etc.
  
  They are persisted into the database and then re-hydrated into a
  "triggerer" process, where many are run at once. We model it so that
  there is a many-to-one relationship between Task and Trigger, for future
  deduplication logic to use.
  
  Rows will be evicted from the database when the triggerer detects no
  active Tasks/DAGs using them. Events are not stored in the database;
  when an Event is fired, the triggerer will directly push its data to the
  appropriate Task/DAG.
class_docstr_tok: ["Base","Trigger","class",".","Triggers","are","a","workload","that","run","in","an","asynchronous","event","loop","shared","with","other","Triggers",",","and","fire","off","events","that","will","unpause","deferred","Tasks",",","start","linked","DAGs",",","etc",".","They","are","persisted","into","the","database","and","then","re-hydrated","into","a","''","triggerer","''","process",",","where","many","are","run","at","once",".","We","model","it","so","that","there","is","a","many-to-one","relationship","between","Task","and","Trigger",",","for","future","deduplication","logic","to","use",".","Rows","will","be","evicted","from","the","database","when","the","triggerer","detects","no","active","Tasks\/DAGs","using","them",".","Events","are","not","stored","in","the","database",";","when","an","Event","is","fired",",","the","triggerer","will","directly","push","its","data","to","the","appropriate","Task\/DAG","."]
docstr: Delete all triggers that have no tasks dependent on them.
  
  Triggers have a one-to-many relationship to task instances, so we need
  to clean those up first. Afterwards we can drop the triggers not
  referenced by anyone.
docstr_tok: ["Delete","all","triggers","that","have","no","tasks","dependent","on","them",".","Triggers","have","a","one-to-many","relationship","to","task","instances",",","so","we","need","to","clean","those","up","first",".","Afterwards","we","can","drop","the","triggers","not","referenced","by","anyone","."]
code: def clean_unused(cls, session: Session = NEW_SESSION) -> None:
      """Delete all triggers that have no tasks dependent on them.
  
      Triggers have a one-to-many relationship to task instances, so we need
      to clean those up first. Afterwards we can drop the triggers not
      referenced by anyone.
      """
      # Update all task instances with trigger IDs that are not DEFERRED to remove them
      for attempt in run_with_db_retries():
          with attempt:
              session.execute(
                  update(TaskInstance)
                  .where(
                      TaskInstance.state != TaskInstanceState.DEFERRED,
                      TaskInstance.trigger_id.is_not(None),
                  )
                  .values(trigger_id=None)
              )
  
      # Get all triggers that have no task instances depending on them...
      ids = session.scalars(
          select(cls.id)
          .join(
              TaskInstance,
              cls.id == TaskInstance.trigger_id,
              isouter=True,
          )
          .group_by(cls.id)
          .having(func.count(TaskInstance.trigger_id) == 0)
      ).all()
      # ...and delete them (we can't do this in one query due to MySQL)
      session.execute(
          delete(Trigger)
          .where(Trigger.id.in_(ids))
          .execution_options(synchronize_session=False)
      )
  
imports: from __future__ import annotations
  import datetime
  from traceback import format_exception
  from typing import TYPE_CHECKING, Any, Iterable
  from sqlalchemy import Column, Integer, String, delete, func, or_, select, update
  from sqlalchemy.orm import joinedload, relationship
  from sqlalchemy.sql.functions import coalesce
  from airflow.api_internal.internal_api_call import internal_api_call
  from airflow.models.base import Base
  from airflow.models.taskinstance import TaskInstance
  from airflow.utils import timezone
  from airflow.utils.retries import run_with_db_retries
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.sqlalchemy import ExtendedJSON, UtcDateTime, with_row_locks
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 4 for calling the Trigger obj's underlying member method code with example usage: obj.clean_unused(cls, session) without return types