Used in https://github.com/mhenc/airflow/blob/master/airflow/dag_processing/manager.py#L613

should be straighforward
"""
        Triggers the appropriate callback depending on the value of success, namely the
        on_failure_callback or on_success_callback. This method gets the context of a
        single TaskInstance part of this DagRun and passes that to the callable along
        with a 'reason', primarily to differentiate DagRun failures.
        .. note: The logs end up in
            ``$AIRFLOW_HOME/logs/scheduler/latest/PROJECT/DAG_FILE.py.log``
        :param dagrun: DagRun object
        :param success: Flag to specify if failure or success callback should be called
        :param reason: Completion reason
        :param session: Database session
        """
        callbacks = self.on_success_callback if success else self.on_failure_callback
        if callbacks:
            callbacks = callbacks if isinstance(callbacks, list) else [callbacks]
            tis = dagrun.get_task_instances(session=session)
            ti = tis[-1]  
            ti.task = self.get_task(ti.task_id)
            context = ti.get_template_context(session=session)
            context.update({"reason": reason})
            for callback in callbacks:
                self.log.info("Executing dag callback function: %s", callback)
                try:
                    callback(context)
                except Exception:
                    self.log.exception("failed to invoke dag state update callback")
                    Stats.incr(
                        "dag.callback_exceptions", tags={"dag_id": dagrun.dag_id, "run_id": dagrun.run_id}
                    )
    def get_active_runs(self):