Replace external_trigger check with DagRunType

    - Fix test_handle_multiple_columns_unique_constraint_error

    - Resolve ashb code review

    - Add newsfragment

Author: jason810496

airflow/api/client/local_client.py

@@ -54,7 +54,6 @@ def trigger_dag(
                 "data_interval_start": dag_run.data_interval_start,
                 "data_interval_end": dag_run.data_interval_end,
                 "end_date": dag_run.end_date,
-                "external_trigger": dag_run.external_trigger,
                 "last_scheduling_decision": dag_run.last_scheduling_decision,
                 "logical_date": dag_run.logical_date,
                 "run_type": dag_run.run_type,

airflow/api/common/mark_tasks.py

@@ -141,7 +141,7 @@ def find_task_relatives(tasks, downstream, upstream):
 @provide_session
 def get_run_ids(dag: DAG, run_id: str, future: bool, past: bool, session: SASession = NEW_SESSION):
     """Return DAG executions' run_ids."""
-    last_dagrun = dag.get_last_dagrun(include_externally_triggered=True, session=session)
+    last_dagrun = dag.get_last_dagrun(include_manually_triggered=True, session=session)
     current_dagrun = dag.get_dagrun(run_id=run_id, session=session)
     first_dagrun = session.scalar(
         select(DagRun).filter(DagRun.dag_id == dag.dag_id).order_by(DagRun.logical_date.asc()).limit(1)

airflow/api/common/trigger_dag.py

@@ -105,7 +105,6 @@ def _trigger_dag(
         conf=run_conf,
         run_type=DagRunType.MANUAL,
         triggered_by=triggered_by,
-        external_trigger=True,
         dag_version=dag_version,
         state=DagRunState.QUEUED,
         session=session,

airflow/api_connexion/endpoints/dag_run_endpoint.py

@@ -186,7 +186,6 @@ def _fetch_dag_runs(
         "start_date",
         "end_date",
         "updated_at",
-        "external_trigger",
         "conf",
     ]
     query = apply_sorting(query, order_by, to_replace, allowed_sort_attrs)
@@ -354,7 +353,6 @@ def post_dag_run(*, dag_id: str, session: Session = NEW_SESSION) -> APIResponse:
                 conf=post_body.get("conf"),
                 run_type=DagRunType.MANUAL,
                 triggered_by=DagRunTriggeredByType.REST_API,
-                external_trigger=True,
                 dag_version=DagVersion.get_latest_version(dag.dag_id),
                 state=DagRunState.QUEUED,
                 session=session,

airflow/api_connexion/schemas/dag_run_schema.py

@@ -67,7 +67,6 @@ class Meta:
     start_date = auto_field(dump_only=True)
     end_date = auto_field(dump_only=True)
     state = DagStateField(dump_only=True)
-    external_trigger = auto_field(dump_default=True, dump_only=True)
     conf = ConfObject()
     data_interval_start = auto_field(validate=validate_istimezone)
     data_interval_end = auto_field(validate=validate_istimezone)

airflow/api_fastapi/core_api/datamodels/dag_run.py

@@ -65,7 +65,6 @@ class DAGRunResponse(BaseModel):
     last_scheduling_decision: datetime | None
     run_type: DagRunType
     state: DagRunState
-    external_trigger: bool
     triggered_by: DagRunTriggeredByType
     conf: dict
     note: str | None

airflow/api_fastapi/core_api/openapi/v1-generated.yaml

@@ -7988,9 +7988,6 @@ components:
           $ref: '#/components/schemas/DagRunType'
         state:
           $ref: '#/components/schemas/DagRunState'
-        external_trigger:
-          type: boolean
-          title: External Trigger
         triggered_by:
           $ref: '#/components/schemas/DagRunTriggeredByType'
         conf:
@@ -8014,7 +8011,6 @@ components:
       - last_scheduling_decision
       - run_type
       - state
-      - external_trigger
       - triggered_by
       - conf
       - note

airflow/api_fastapi/core_api/routes/public/dag_run.py

@@ -283,7 +283,6 @@ def get_dag_runs(
                     "start_date",
                     "end_date",
                     "updated_at",
-                    "external_trigger",
                     "conf",
                 ],
                 DagRun,
@@ -380,7 +379,6 @@ def trigger_dag_run(
             conf=body.conf,
             run_type=DagRunType.MANUAL,
             triggered_by=DagRunTriggeredByType.REST_API,
-            external_trigger=True,
             dag_version=DagVersion.get_latest_version(dag.dag_id),
             state=DagRunState.QUEUED,
             session=session,
@@ -429,7 +427,6 @@ def get_list_dag_runs_batch(
             "start_date",
             "end_date",
             "updated_at",
-            "external_trigger",
             "conf",
         ],
         DagRun,

airflow/api_fastapi/execution_api/datamodels/taskinstance.py

@@ -223,7 +223,6 @@ class DagRun(BaseModel):
     end_date: UtcDateTime | None
     run_type: DagRunType
     conf: Annotated[dict[str, Any], Field(default_factory=dict)]
-    external_trigger: bool = False
 
 
 class TIRunContext(BaseModel):

airflow/cli/commands/remote_commands/task_command.py

@@ -174,7 +174,6 @@ def _get_dag_run(
             dag_id=dag.dag_id,
             run_id=logical_date_or_run_id,
             run_type=DagRunType.MANUAL,
-            external_trigger=True,
             logical_date=dag_run_logical_date,
             data_interval=dag.timetable.infer_manual_data_interval(run_after=dag_run_logical_date),
             triggered_by=DagRunTriggeredByType.CLI,