"""
        Deletes all triggers that have no tasks/DAGs dependent on them
        (triggers have a one-to-many relationship to both)
        """
        
        for attempt in run_with_db_retries():
            with attempt:
                session.query(TaskInstance).filter(
                    TaskInstance.state != State.DEFERRED, TaskInstance.trigger_id.isnot(None)
                ).update({TaskInstance.trigger_id: None})
        
        ids = [
            trigger_id
            for (trigger_id,) in (
                session.query(cls.id)
                .join(TaskInstance, cls.id == TaskInstance.trigger_id, isouter=True)
                .group_by(cls.id)
                .having(func.count(TaskInstance.trigger_id) == 0)
            )
        ]
        
        session.query(Trigger).filter(Trigger.id.in_(ids)).delete(synchronize_session=False)
    @classmethod
    @provide_session
    def submit_event(cls, trigger_id, event, session=None):