ExternalTaskSensor External link with incorrect URL #61850

def get_last_execution_date_of_dag(logical_date: datetime.datetime, **kwargs):
    import airflow_client.client

    from airflow.configuration import conf
    from myprovider.hooks.api_token import AirflowAPITokenHook  # Won't work, you need to obtain the token somehow

    hook = AirflowAPITokenHook(airflow_api_token_conn_id="airflow_api_token_default")
    token = hook.get_token()

    base_url = conf.get("api", "base_url")

    dag_id = kwargs["task"].external_dag_id

    configuration = airflow_client.client.Configuration(
        host=base_url, access_token=token
    )
    with airflow_client.client.ApiClient(configuration) as api_client:
        api_instance = airflow_client.client.DagRunApi(api_client)

        api_response = api_instance.get_dag_runs(
            dag_id, logical_date_lte=logical_date, limit=1, order_by=["-logical_date"]
        )

        if not api_response.dag_runs or len(api_response.dag_runs) == 0:
            raise ValueError(
                f"No previous dag_run found for DAG {dag_id} before {logical_date}. Please ensure the parent Dag has at least one run before this date."
            )

        last_dag_run = api_response.dag_runs[0]
        parent_logical_date = last_dag_run.logical_date

        print(
            f"Found last dag_run for DAG {dag_id} before {logical_date}: {parent_logical_date}"
        )

    return parent_logical_date