o-nikolas commented on code in PR #62984:
URL: https://github.com/apache/airflow/pull/62984#discussion_r2921507643
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -127,26 +128,40 @@ def __init__(self, *args, **kwargs):
def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
from airflow.executors import workloads
- if not isinstance(workload, workloads.ExecuteTask):
+ if AIRFLOW_V_3_2_PLUS and isinstance(workload,
workloads.ExecuteCallback):
+ self.queued_callbacks[workload.callback.id] = workload
+ elif isinstance(workload, workloads.ExecuteTask):
+ ti = workload.ti
+ self.queued_tasks[ti.key] = workload
+ else:
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: Sequence[workloads.All]) -> None:
- from airflow.executors.workloads import ExecuteTask
+ from airflow.executors import workloads as wl
# Airflow V3 version
for w in workloads:
- if not isinstance(w, ExecuteTask):
+ if isinstance(w, wl.ExecuteTask):
+ tcommand = [w]
Review Comment:
Please use full names, the clarity is worth 4 more characters :)
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -447,6 +466,9 @@ def _submit_job_kwargs(
submit_job_api["containerOverrides"]["environment"].append(
{"name": "AIRFLOW_IS_EXECUTOR_CONTAINER", "value": "true"}
)
+ if queue:
+ submit_job_api["jobQueue"] = queue
Review Comment:
Let's not conflate this change along with supporting callbacks. So far this
executor has not been opinionated about the `queue` parameter. I like your
suggestion, but let's discuss it in it's own PR.
##########
providers/amazon/src/airflow/providers/amazon/aws/executors/batch/batch_executor.py:
##########
@@ -127,26 +128,40 @@ def __init__(self, *args, **kwargs):
def queue_workload(self, workload: workloads.All, session: Session | None)
-> None:
Review Comment:
If you don't mind can you narrow down the type for workload here? I think
there is a new type that @ferruzzi made basically anywhere you need `task |
callback` as a type.
--
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]
