o-nikolas commented on code in PR #48513:
URL: https://github.com/apache/airflow/pull/48513#discussion_r2027752631
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/ecs/ecs_executor.py:
##########
@@ -114,6 +123,32 @@ def __init__(self, *args, **kwargs):
self.run_task_kwargs = self._load_run_kwargs()
+ def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
+ from airflow.executors import workloads
+
+ if not isinstance(workload, workloads.ExecuteTask):
+ raise RuntimeError(f"{type(self)} cannot handle workloads of type
{type(workload)}")
+ ti = workload.ti
+ self.queued_tasks[ti.key] = workload
+
+ def _process_workloads(self, workloads: list[workloads.All]) -> None:
+ from airflow.executors.workloads import ExecuteTask
+
+ # Airflow V3 version
+ for w in workloads:
+ if not isinstance(w, ExecuteTask):
+ raise RuntimeError(f"{type(self)} cannot handle workloads of
type {type(w)}")
+
+ # TODO: AIP-72 handle populating tokens once
https://github.com/apache/airflow/issues/45107 is handled.
+ command = [w]
+ key = w.ti.key
Review Comment:
Fair, must have happened in main while I was branched. I'll drop that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
