Repository: apache/airflow
File: providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py (Python)

Code before:
        else:
            self._trigger_dag_af_2(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )

    def _trigger_dag_af_3(self, context, run_id, parsed_logical_date):
        from airflow.providers.common.compat.sdk import DagRunTriggerException

        kwargs_accepted = dict(
            trigger_dag_id=self.trigger_dag_id,
            dag_run_id=run_id,
            conf=self.conf,
            logical_date=parsed_logical_date,
            reset_dag_run=self.reset_dag_run,
            skip_when_already_exists=self.skip_when_already_exists,
            wait_for_completion=self.wait_for_completion,
            allowed_states=self.allowed_states,
            failed_states=self.failed_states,
            poke_interval=self.poke_interval,
            deferrable=self.deferrable,
        )

        if self.note:
            sig = inspect.signature(DagRunTriggerException.__init__)
            if "note" in sig.parameters:
                kwargs_accepted["note"] = self.note

        raise DagRunTriggerException(**kwargs_accepted)

    def _trigger_dag_af_2(self, context, run_id, parsed_logical_date):
        try:
            dag_run = trigger_dag(
                dag_id=self.trigger_dag_id,
                run_id=run_id,
                conf=self.conf,
                execution_date=parsed_logical_date,
                replace_microseconds=False,
            )

        except DagRunAlreadyExists as e:
            if self.reset_dag_run:
                dag_run = e.dag_run
                self.log.info("Clearing %s on %s", self.trigger_dag_id, dag_run.run_id)

                # Get target dag object and call clear()
                dag_model = DagModel.get_current(self.trigger_dag_id)
                if dag_model is None:
                    raise DagNotFound(f"Dag id {self.trigger_dag_id} not found in DagModel")

                # Note: here execution fails on database isolation mode. Needs structural changes for AIP-72
                dag = SerializedDagModel.get_dag(self.trigger_dag_id)

Review comment (suggestion):
```suggestion
        if self.note and "note" in inspect.signature(DagRunTriggerException.__init__).parameters:
            kwargs_accepted["note"] = self.note
```

Code after:
        else:
            self._trigger_dag_af_2(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )

    def _trigger_dag_af_3(self, context, run_id, parsed_logical_date):
        from airflow.providers.common.compat.sdk import DagRunTriggerException

        kwargs_accepted = dict(
            trigger_dag_id=self.trigger_dag_id,
            dag_run_id=run_id,
            conf=self.conf,
            logical_date=parsed_logical_date,
            reset_dag_run=self.reset_dag_run,
            skip_when_already_exists=self.skip_when_already_exists,
            wait_for_completion=self.wait_for_completion,
            allowed_states=self.allowed_states,
            failed_states=self.failed_states,
            poke_interval=self.poke_interval,
            deferrable=self.deferrable,
        )

        if self.note and "note" in inspect.signature(DagRunTriggerException.__init__).parameters:
            kwargs_accepted["note"] = self.note

        raise DagRunTriggerException(**kwargs_accepted)

    def _trigger_dag_af_2(self, context, run_id, parsed_logical_date):
        try:
            if self.note:
                self.log.warning("Parameter 'note' is not supported in Airflow 2.x and will be ignored.")

            dag_run = trigger_dag(
                dag_id=self.trigger_dag_id,
                run_id=run_id,
                conf=self.conf,
                execution_date=parsed_logical_date,
                replace_microseconds=False,
            )

        except DagRunAlreadyExists as e:
            if self.reset_dag_run:
                dag_run = e.dag_run
                self.log.info("Clearing %s on %s", self.trigger_dag_id, dag_run.run_id)

                # Get target dag object and call clear()
                dag_model = DagModel.get_current(self.trigger_dag_id)
                if dag_model is None:
                    raise DagNotFound(f"Dag id {self.trigger_dag_id} not found in DagModel")

                # Note: here execution fails on database isolation mode. Needs structural changes for AIP-72