Raise error upon DAG import instead of DAG trigger for incorrect pool_slots value in mapped tasks using partial()

airflow/decorators/base.py

@@ -457,6 +457,11 @@ def _expand(self, expand_input: ExpandInput, *, strict: bool) -> XComArg:
         end_date = timezone.convert_to_utc(partial_kwargs.pop("end_date", None))
         if partial_kwargs.get("pool") is None:
             partial_kwargs["pool"] = Pool.DEFAULT_POOL_NAME
+        if partial_kwargs.get("pool_slots") and partial_kwargs["pool_slots"] < 1:
+            dag_str = ""
+            if partial_kwargs.get("dag"):
+                dag_str = f" in dag {partial_kwargs['dag'].dag_id}"
+            raise ValueError(f"pool slots for {task_id}{dag_str} cannot be less than 1")
         partial_kwargs["retries"] = parse_retries(partial_kwargs.get("retries", DEFAULT_RETRIES))
         partial_kwargs["retry_delay"] = coerce_timedelta(
             partial_kwargs.get("retry_delay", DEFAULT_RETRY_DELAY),

airflow/models/baseoperator.py

@@ -362,6 +362,11 @@ def partial(
     partial_kwargs["end_date"] = timezone.convert_to_utc(partial_kwargs["end_date"])
     if partial_kwargs["pool"] is None:
         partial_kwargs["pool"] = Pool.DEFAULT_POOL_NAME
+    if partial_kwargs["pool_slots"] < 1:
+        dag_str = ""
+        if partial_kwargs["dag"]:
+            dag_str = f" in dag {partial_kwargs['dag'].dag_id}"
+        raise ValueError(f"pool slots for {partial_kwargs['task_id']}{dag_str} cannot be less than 1")
     partial_kwargs["retries"] = parse_retries(partial_kwargs["retries"])
     partial_kwargs["retry_delay"] = coerce_timedelta(partial_kwargs["retry_delay"], key="retry_delay")
     if partial_kwargs["max_retry_delay"] is not None:

tests/models/test_mappedoperator.py

@@ -218,6 +218,15 @@ def test_partial_on_class_invalid_ctor_args() -> None:
         MockOperator.partial(task_id="a", foo="bar", bar=2)
 
 
+def test_partial_on_invalid_pool_slots_raises() -> None:
+    """Test that when we pass an invalid value to pool_slots in partial(),
+
+    i.e. if the value is not an integer, an error is raised at import time."""
+
+    with pytest.raises(TypeError, match="'<' not supported between instances of 'str' and 'int'"):
+        MockOperator.partial(task_id="pool_slots_test", pool="test", pool_slots="a").expand(arg1=[1, 2, 3])
+
+
 @pytest.mark.parametrize(
     ["num_existing_tis", "expected"],
     (