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

Code before:
        skip_when_already_exists: bool = False,
        fail_when_dag_is_paused: bool = False,
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        openlineage_inject_parent_info: bool = True,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.trigger_dag_id = trigger_dag_id
        self.trigger_run_id = trigger_run_id
        self.conf = conf
        self.reset_dag_run = reset_dag_run
        self.wait_for_completion = wait_for_completion
        self.poke_interval = poke_interval
        if allowed_states:
            self.allowed_states = [DagRunState(s) for s in allowed_states]
        else:
            self.allowed_states = [DagRunState.SUCCESS]
        if failed_states is not None:
            self.failed_states = [DagRunState(s) for s in failed_states]
        else:
            self.failed_states = [DagRunState.FAILED]
        self.skip_when_already_exists = skip_when_already_exists
        self.fail_when_dag_is_paused = fail_when_dag_is_paused
        self.openlineage_inject_parent_info = openlineage_inject_parent_info
        # deferrable only makes sense when wait_for_completion=True
        self.deferrable = deferrable and wait_for_completion
        self.logical_date = logical_date
        if logical_date is NOTSET:
            self.logical_date = NOTSET
        elif logical_date is None or isinstance(logical_date, (str, datetime.datetime)):
            self.logical_date = logical_date
        else:
            raise TypeError(
                f"Expected str, datetime.datetime, or None for parameter 'logical_date'. Got {type(logical_date).__name__}"
            )

        if fail_when_dag_is_paused and AIRFLOW_V_3_0_PLUS:
            raise NotImplementedError("Setting `fail_when_dag_is_paused` not yet supported for Airflow 3.x")

    def execute(self, context: Context):
        if self.logical_date is NOTSET:
            # If no logical_date is provided we will set utcnow()
            parsed_logical_date = timezone.utcnow()
        elif self.logical_date is None or isinstance(self.logical_date, datetime.datetime):
            parsed_logical_date = self.logical_date  # type: ignore
        elif isinstance(self.logical_date, str):
            parsed_logical_date = timezone.parse(self.logical_date)

        try:
            if self.conf and isinstance(self.conf, str):
                self.conf = json.loads(self.conf)

Review comment (suggestion):
If it only makes sense when `wait_for_completion = True`:
1. It should raise an exception when it's not (inc. reasoning + suggestion for fixing actions) 
2. The exception should be tested
3. It should be documented in the docstring (for both parameters)

Code after:
        skip_when_already_exists: bool = False,
        fail_when_dag_is_paused: bool = False,
        deferrable: bool = conf.getboolean("operators", "default_deferrable", fallback=False),
        openlineage_inject_parent_info: bool = True,
        **kwargs,
    ) -> None:
        super().__init__(**kwargs)
        self.trigger_dag_id = trigger_dag_id
        self.trigger_run_id = trigger_run_id
        self.conf = conf
        self.reset_dag_run = reset_dag_run
        self.wait_for_completion = wait_for_completion
        self.poke_interval = poke_interval
        if allowed_states:
            self.allowed_states = [DagRunState(s) for s in allowed_states]
        else:
            self.allowed_states = [DagRunState.SUCCESS]
        if failed_states is not None:
            self.failed_states = [DagRunState(s) for s in failed_states]
        else:
            self.failed_states = [DagRunState.FAILED]
        self.skip_when_already_exists = skip_when_already_exists
        self.fail_when_dag_is_paused = fail_when_dag_is_paused
        self.openlineage_inject_parent_info = openlineage_inject_parent_info
        self.deferrable = deferrable
        self.logical_date = logical_date
        if logical_date is NOTSET:
            self.logical_date = NOTSET
        elif logical_date is None or isinstance(logical_date, (str, datetime.datetime)):
            self.logical_date = logical_date
        else:
            raise TypeError(
                f"Expected str, datetime.datetime, or None for parameter 'logical_date'. Got {type(logical_date).__name__}"
            )

        if fail_when_dag_is_paused and AIRFLOW_V_3_0_PLUS:
            raise NotImplementedError("Setting `fail_when_dag_is_paused` not yet supported for Airflow 3.x")

    def execute(self, context: Context):
        if self.logical_date is NOTSET:
            # If no logical_date is provided we will set utcnow()
            parsed_logical_date = timezone.utcnow()
        elif self.logical_date is None or isinstance(self.logical_date, datetime.datetime):
            parsed_logical_date = self.logical_date  # type: ignore
        elif isinstance(self.logical_date, str):
            parsed_logical_date = timezone.parse(self.logical_date)

        try:
            if self.conf and isinstance(self.conf, str):
                self.conf = json.loads(self.conf)
            json.dumps(self.conf)