Validate task_id for mapped tasks

A regular task validates its task_id in BaseOperator.__init__, but a
mapped task is built through a different path that skips __init__, so its
task_id was never checked. An invalid id such as "bad*id" passed
silently in .partial().expand() and in the taskflow @task mapped form,
only to fail later where the id is used. Validate the id in both mapped
paths so it is rejected at parse time like a regular task.

Author: Andrushika

task-sdk/src/airflow/sdk/bases/decorator.py

@@ -45,6 +45,7 @@
     ListOfDictsExpandInput,
     is_mappable,
 )
+from airflow.sdk.definitions._internal.node import validate_key
 from airflow.sdk.definitions._internal.types import NOTSET
 from airflow.sdk.definitions.asset import Asset
 from airflow.sdk.definitions.context import KNOWN_CONTEXT_KEYS
@@ -621,6 +622,9 @@ def _expand(self, expand_input: ExpandInput, *, strict: bool) -> XComArg:
         task_id = get_unique_task_id(partial_kwargs.pop("task_id"), dag, task_group)
         if task_group:
             task_id = task_group.child_id(task_id)
+        # Mapped tasks skip BaseOperator.__init__, so validate the id here to match a
+        # regular task (see BaseOperator.__init__).
+        validate_key(task_id)
 
         # Logic here should be kept in sync with BaseOperatorMeta.partial().
         if partial_kwargs.get("wait_for_downstream"):

task-sdk/src/airflow/sdk/bases/operator.py

@@ -345,6 +345,9 @@ def partial(
             task_group = task_group or TaskGroupContext.get_current(dag)
         if task_group:
             task_id = task_group.child_id(task_id)
+        # Mapped tasks skip BaseOperator.__init__, so validate the id here to match a
+        # regular task (see BaseOperator.__init__).
+        validate_key(task_id)
 
         # Merge Dag and task group level defaults into user-supplied values.
         dag_default_args, partial_params = get_merged_defaults(

task-sdk/tests/task_sdk/definitions/decorators/test_mapped.py

@@ -17,9 +17,22 @@
 # under the License.
 from __future__ import annotations
 
+import pytest
+
 from airflow.sdk import DAG, TaskGroup
 
 
+def test_mapped_taskflow_task_id_is_validated():
+    """A mapped @task skips BaseOperator.__init__, so its id must still be validated."""
+
+    def f(z):
+        pass
+
+    with DAG(dag_id="d", schedule=None) as dag:
+        with pytest.raises(ValueError, match="has to be made of alphanumeric"):
+            dag.task(task_id="bad*id")(f).expand(z=[])
+
+
 def test_mapped_task_group_id_prefix_task_id():
     def f(z):
         pass

task-sdk/tests/task_sdk/definitions/test_mappedoperator.py

@@ -75,6 +75,21 @@ def test_task_mapping_with_dag():
 # test_task_mapping_with_dag_and_list_of_pandas_dataframe
 
 
+@pytest.mark.parametrize(
+    ("bad_task_id", "match"),
+    [
+        ("bad*id", "has to be made of alphanumeric"),
+        ("with space", "has to be made of alphanumeric"),
+        ("a" * 251, "has to be less than 250 characters"),
+    ],
+)
+def test_mapped_task_id_is_validated(bad_task_id, match):
+    """A mapped task skips BaseOperator.__init__, so its id must still be validated."""
+    with DAG("test-dag"):
+        with pytest.raises(ValueError, match=match):
+            MockOperator.partial(task_id=bad_task_id).expand(arg2=["a"])
+
+
 def test_task_mapping_without_dag_context():
     with DAG("test-dag") as dag:
         task1 = BaseOperator(task_id="op1")