<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0</p>
<h3 dir="auto">What happened</h3>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="from datetime import datetime

from airflow import DAG
from airflow.decorators import task
from typing import List, Dict
from airflow.operators.dummy import DummyOperator
from airflow.operators.python import PythonOperator
from airflow.utils.trigger_rule import TriggerRule

def end_function(name):
    print(f&quot;Calling end {name}&quot;)

with DAG(dag_id=&quot;not_simple_mapping&quot;, start_date=datetime(2022, 6, 7), schedule_interval=None) as dag:

    @task()
    def show(x: dict):
        print(f&quot;x={x}&quot;)

    x_list: List[Dict] = [{}]
    end_list: list = []

    for i in range(0, 5):
        x_list.append({
            'i': i,
            'i_plus_one': i+1
        })
        end_list.append(f'Task{i}')

    start_region_job_task = DummyOperator(
        task_id='Start_Region',
        trigger_rule=TriggerRule.ALL_DONE
    )
    
    main_run = show.expand(x=x_list)
    end_run = PythonOperator.partial(task_id='End_Region', python_callable=end_function, do_xcom_push=False).expand(
        op_kwargs = [{'name': 'Task1'}, {'name': 'Task2'}]
    )

    start_region_job_task &gt;&gt; main_run &gt;&gt; end_run
    "><pre class="notranslate"><span class="pl-k">from</span> <span class="pl-s1">datetime</span> <span class="pl-k">import</span> <span class="pl-s1">datetime</span>

<span class="pl-k">from</span> <span class="pl-s1">airflow</span> <span class="pl-k">import</span> <span class="pl-v">DAG</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">task</span>
<span class="pl-k">from</span> <span class="pl-s1">typing</span> <span class="pl-k">import</span> <span class="pl-v">List</span>, <span class="pl-v">Dict</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">dummy</span> <span class="pl-k">import</span> <span class="pl-v">DummyOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">python</span> <span class="pl-k">import</span> <span class="pl-v">PythonOperator</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">trigger_rule</span> <span class="pl-k">import</span> <span class="pl-v">TriggerRule</span>

<span class="pl-k">def</span> <span class="pl-en">end_function</span>(<span class="pl-s1">name</span>):
    <span class="pl-en">print</span>(<span class="pl-s">f"Calling end <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">name</span><span class="pl-kos">}</span></span>"</span>)

<span class="pl-k">with</span> <span class="pl-v">DAG</span>(<span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">"not_simple_mapping"</span>, <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">datetime</span>(<span class="pl-c1">2022</span>, <span class="pl-c1">6</span>, <span class="pl-c1">7</span>), <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>) <span class="pl-k">as</span> <span class="pl-s1">dag</span>:

    <span class="pl-en">@<span class="pl-en">task</span>()</span>
    <span class="pl-k">def</span> <span class="pl-en">show</span>(<span class="pl-s1">x</span>: <span class="pl-s1">dict</span>):
        <span class="pl-en">print</span>(<span class="pl-s">f"x=<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">x</span><span class="pl-kos">}</span></span>"</span>)

    <span class="pl-s1">x_list</span>: <span class="pl-v">List</span>[<span class="pl-v">Dict</span>] <span class="pl-c1">=</span> [{}]
    <span class="pl-s1">end_list</span>: <span class="pl-s1">list</span> <span class="pl-c1">=</span> []

    <span class="pl-k">for</span> <span class="pl-s1">i</span> <span class="pl-c1">in</span> <span class="pl-en">range</span>(<span class="pl-c1">0</span>, <span class="pl-c1">5</span>):
        <span class="pl-s1">x_list</span>.<span class="pl-en">append</span>({
            <span class="pl-s">'i'</span>: <span class="pl-s1">i</span>,
            <span class="pl-s">'i_plus_one'</span>: <span class="pl-s1">i</span><span class="pl-c1">+</span><span class="pl-c1">1</span>
        })
        <span class="pl-s1">end_list</span>.<span class="pl-en">append</span>(<span class="pl-s">f'Task<span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">i</span><span class="pl-kos">}</span></span>'</span>)

    <span class="pl-s1">start_region_job_task</span> <span class="pl-c1">=</span> <span class="pl-v">DummyOperator</span>(
        <span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'Start_Region'</span>,
        <span class="pl-s1">trigger_rule</span><span class="pl-c1">=</span><span class="pl-v">TriggerRule</span>.<span class="pl-v">ALL_DONE</span>
    )
    
    <span class="pl-s1">main_run</span> <span class="pl-c1">=</span> <span class="pl-s1">show</span>.<span class="pl-en">expand</span>(<span class="pl-s1">x</span><span class="pl-c1">=</span><span class="pl-s1">x_list</span>)
    <span class="pl-s1">end_run</span> <span class="pl-c1">=</span> <span class="pl-v">PythonOperator</span>.<span class="pl-en">partial</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">'End_Region'</span>, <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-s1">end_function</span>, <span class="pl-s1">do_xcom_push</span><span class="pl-c1">=</span><span class="pl-c1">False</span>).<span class="pl-en">expand</span>(
        <span class="pl-s1">op_kwargs</span> <span class="pl-c1">=</span> [{<span class="pl-s">'name'</span>: <span class="pl-s">'Task1'</span>}, {<span class="pl-s">'name'</span>: <span class="pl-s">'Task2'</span>}]
    )

    <span class="pl-s1">start_region_job_task</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">main_run</span> <span class="pl-c1">&gt;&gt;</span> <span class="pl-s1">end_run</span>
    </pre></div>
<p dir="auto">Could not works because of passing list of kwargs</p>
<div class="highlight highlight-source-shell notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="  File &quot;/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py&quot;, line 343, in _serialize
    return SerializedBaseOperator.serialize_mapped_operator(var)
  File &quot;/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py&quot;, line 608, in serialize_mapped_operator
    assert op_kwargs[Encoding.TYPE] == DAT.DICT
TypeError: list indices must be integers or slices, not Encoding

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File &quot;/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py&quot;, line 1105, in to_dict
    json_dict = {&quot;__version&quot;: cls.SERIALIZER_VERSION, &quot;dag&quot;: cls.serialize_dag(var)}
  File &quot;/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py&quot;, line 1013, in serialize_dag
    raise SerializationError(f'Failed to serialize DAG {dag.dag_id!r}: {e}')
airflow.exceptions.SerializationError: Failed to serialize DAG 'not_simple_mapping': list indices must be integers or slices, not Encoding"><pre class="notranslate">  File <span class="pl-s"><span class="pl-pds">"</span>/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py<span class="pl-pds">"</span></span>, line 343, <span class="pl-k">in</span> _serialize
    <span class="pl-k">return</span> SerializedBaseOperator.serialize_mapped_operator(var)
  File <span class="pl-s"><span class="pl-pds">"</span>/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py<span class="pl-pds">"</span></span>, line 608, <span class="pl-k">in</span> serialize_mapped_operator
    assert op_kwargs[Encoding.TYPE] == DAT.DICT
TypeError: list indices must be integers or slices, not Encoding

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File <span class="pl-s"><span class="pl-pds">"</span>/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py<span class="pl-pds">"</span></span>, line 1105, <span class="pl-k">in</span> to_dict
    json_dict = {<span class="pl-s"><span class="pl-pds">"</span>__version<span class="pl-pds">"</span></span>: cls.SERIALIZER_VERSION, <span class="pl-s"><span class="pl-pds">"</span>dag<span class="pl-pds">"</span></span>: cls.serialize_dag(var)}
  File <span class="pl-s"><span class="pl-pds">"</span>/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py<span class="pl-pds">"</span></span>, line 1013, <span class="pl-k">in</span> serialize_dag
    raise SerializationError(f<span class="pl-s"><span class="pl-pds">'</span>Failed to serialize DAG {dag.dag_id!r}: {e}<span class="pl-pds">'</span></span>)
airflow.exceptions.SerializationError: Failed to serialize DAG <span class="pl-s"><span class="pl-pds">'</span>not_simple_mapping<span class="pl-pds">'</span></span>: list indices must be integers or slices, not Encoding</pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">It's should accept a list of kwargs to trigger dynamic mapping instead of one kwargs</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Ubuntu 22.04</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Virtualenv installation</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>
<h3 dir="auto">Apache Airflow version</h3>
<p dir="auto">2.3.0 (latest released)</p>
<h3 dir="auto">What happened</h3>
<p dir="auto">The following DAG was written and expected to generate 3 tasks (one for each string in the list)</p>
<p dir="auto"><strong>dag_code</strong></p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="import logging
from airflow.decorators import dag, task
from airflow.operators.python import PythonOperator

from airflow.utils.dates import datetime


def log_strings_operator(string, *args, **kwargs):
    logging.info(&quot;we've made it into the method&quot;)
    logging.info(f&quot;operator log - {string}&quot;)


@dag(
    dag_id='dynamic_dag_test',
    schedule_interval=None,
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=['example', 'dynamic_tasks']
)
def tutorial_taskflow_api_etl():
    op2 = (PythonOperator
           .partial(task_id=&quot;logging_with_operator_task&quot;,
                    python_callable=log_strings_operator)
           .expand(op_kwargs=[{&quot;string&quot;: &quot;a&quot;}, {&quot;string&quot;: &quot;b&quot;}, {&quot;string&quot;: &quot;c&quot;}]))

    return op2


tutorial_etl_dag = tutorial_taskflow_api_etl()"><pre class="notranslate"><span class="pl-k">import</span> <span class="pl-s1">logging</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">decorators</span> <span class="pl-k">import</span> <span class="pl-s1">dag</span>, <span class="pl-s1">task</span>
<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">operators</span>.<span class="pl-s1">python</span> <span class="pl-k">import</span> <span class="pl-v">PythonOperator</span>

<span class="pl-k">from</span> <span class="pl-s1">airflow</span>.<span class="pl-s1">utils</span>.<span class="pl-s1">dates</span> <span class="pl-k">import</span> <span class="pl-s1">datetime</span>


<span class="pl-k">def</span> <span class="pl-en">log_strings_operator</span>(<span class="pl-s1">string</span>, <span class="pl-c1">*</span><span class="pl-s1">args</span>, <span class="pl-c1">**</span><span class="pl-s1">kwargs</span>):
    <span class="pl-s1">logging</span>.<span class="pl-en">info</span>(<span class="pl-s">"we've made it into the method"</span>)
    <span class="pl-s1">logging</span>.<span class="pl-en">info</span>(<span class="pl-s">f"operator log - <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">string</span><span class="pl-kos">}</span></span>"</span>)


<span class="pl-en">@<span class="pl-en">dag</span>(</span>
<span class="pl-en">    <span class="pl-s1">dag_id</span><span class="pl-c1">=</span><span class="pl-s">'dynamic_dag_test'</span>,</span>
<span class="pl-en">    <span class="pl-s1">schedule_interval</span><span class="pl-c1">=</span><span class="pl-c1">None</span>,</span>
<span class="pl-en">    <span class="pl-s1">start_date</span><span class="pl-c1">=</span><span class="pl-en">datetime</span>(<span class="pl-c1">2021</span>, <span class="pl-c1">1</span>, <span class="pl-c1">1</span>),</span>
<span class="pl-en">    <span class="pl-s1">catchup</span><span class="pl-c1">=</span><span class="pl-c1">False</span>,</span>
<span class="pl-en">    <span class="pl-s1">tags</span><span class="pl-c1">=</span>[<span class="pl-s">'example'</span>, <span class="pl-s">'dynamic_tasks'</span>]</span>
<span class="pl-en">)</span>
<span class="pl-k">def</span> <span class="pl-en">tutorial_taskflow_api_etl</span>():
    <span class="pl-s1">op2</span> <span class="pl-c1">=</span> (<span class="pl-v">PythonOperator</span>
           .<span class="pl-en">partial</span>(<span class="pl-s1">task_id</span><span class="pl-c1">=</span><span class="pl-s">"logging_with_operator_task"</span>,
                    <span class="pl-s1">python_callable</span><span class="pl-c1">=</span><span class="pl-s1">log_strings_operator</span>)
           .<span class="pl-en">expand</span>(<span class="pl-s1">op_kwargs</span><span class="pl-c1">=</span>[{<span class="pl-s">"string"</span>: <span class="pl-s">"a"</span>}, {<span class="pl-s">"string"</span>: <span class="pl-s">"b"</span>}, {<span class="pl-s">"string"</span>: <span class="pl-s">"c"</span>}]))

    <span class="pl-k">return</span> <span class="pl-s1">op2</span>


<span class="pl-s1">tutorial_etl_dag</span> <span class="pl-c1">=</span> <span class="pl-en">tutorial_taskflow_api_etl</span>()</pre></div>
<p dir="auto"><strong>error message</strong></p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="Broken DAG: [/usr/local/airflow/dags/dynamic_dag_test.py] Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py&quot;, line 343, in _serialize
    return SerializedBaseOperator.serialize_mapped_operator(var)
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py&quot;, line 608, in serialize_mapped_operator
    assert op_kwargs[Encoding.TYPE] == DAT.DICT
TypeError: list indices must be integers or slices, not Encoding

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py&quot;, line 1105, in to_dict
    json_dict = {&quot;__version&quot;: cls.SERIALIZER_VERSION, &quot;dag&quot;: cls.serialize_dag(var)}
  File &quot;/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py&quot;, line 1013, in serialize_dag
    raise SerializationError(f'Failed to serialize DAG {dag.dag_id!r}: {e}')
airflow.exceptions.SerializationError: Failed to serialize DAG 'dynamic_dag_test': list indices must be integers or slices, not Encoding"><pre class="notranslate"><span class="pl-v">Broken</span> <span class="pl-v">DAG</span>: [<span class="pl-c1">/</span><span class="pl-s1">usr</span><span class="pl-c1">/</span><span class="pl-s1">local</span><span class="pl-c1">/</span><span class="pl-s1">airflow</span><span class="pl-c1">/</span><span class="pl-s1">dags</span><span class="pl-c1">/</span><span class="pl-s1">dynamic_dag_test</span>.<span class="pl-s1">py</span>] <span class="pl-v">Traceback</span> (<span class="pl-s1">most</span> <span class="pl-s1">recent</span> <span class="pl-s1">call</span> <span class="pl-s1">last</span>):
  <span class="pl-v">File</span> <span class="pl-s">"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py"</span>, <span class="pl-s1">line</span> <span class="pl-c1">343</span>, <span class="pl-c1">in</span> <span class="pl-s1">_serialize</span>
    <span class="pl-k">return</span> <span class="pl-v">SerializedBaseOperator</span>.<span class="pl-en">serialize_mapped_operator</span>(<span class="pl-s1">var</span>)
  <span class="pl-v">File</span> <span class="pl-s">"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py"</span>, <span class="pl-s1">line</span> <span class="pl-c1">608</span>, <span class="pl-s1">in</span> <span class="pl-s1">serialize_mapped_operator</span>
    <span class="pl-k">assert</span> <span class="pl-s1">op_kwargs</span>[<span class="pl-v">Encoding</span>.<span class="pl-v">TYPE</span>] <span class="pl-c1">==</span> <span class="pl-v">DAT</span>.<span class="pl-v">DICT</span>
<span class="pl-v">TypeError</span>: <span class="pl-s1">list</span> <span class="pl-s1">indices</span> <span class="pl-s1">must</span> <span class="pl-s1">be</span> <span class="pl-s1">integers</span> <span class="pl-c1">or</span> <span class="pl-s1">slices</span>, <span class="pl-c1">not</span> <span class="pl-v">Encoding</span>

<span class="pl-v">During</span> <span class="pl-s1">handling</span> <span class="pl-s1">of</span> <span class="pl-s1">the</span> <span class="pl-s1">above</span> <span class="pl-s1">exception</span>, <span class="pl-s1">another</span> <span class="pl-s1">exception</span> <span class="pl-s1">occurred</span>:

<span class="pl-v">Traceback</span> (<span class="pl-s1">most</span> <span class="pl-s1">recent</span> <span class="pl-s1">call</span> <span class="pl-s1">last</span>):
  <span class="pl-v">File</span> <span class="pl-s">"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py"</span>, <span class="pl-s1">line</span> <span class="pl-c1">1105</span>, <span class="pl-s1">in</span> <span class="pl-s1">to_dict</span>
    <span class="pl-s1">json_dict</span> <span class="pl-c1">=</span> {<span class="pl-s">"__version"</span>: <span class="pl-s1">cls</span>.<span class="pl-v">SERIALIZER_VERSION</span>, <span class="pl-s">"dag"</span>: <span class="pl-s1">cls</span>.<span class="pl-en">serialize_dag</span>(<span class="pl-s1">var</span>)}
  <span class="pl-v">File</span> <span class="pl-s">"/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py"</span>, <span class="pl-s1">line</span> <span class="pl-c1">1013</span>, <span class="pl-s1">in</span> <span class="pl-s1">serialize_dag</span>
    <span class="pl-k">raise</span> <span class="pl-v">SerializationError</span>(<span class="pl-s">f'Failed to serialize DAG <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">dag</span>.<span class="pl-s1">dag_id</span>!r<span class="pl-kos">}</span></span>: <span class="pl-s1"><span class="pl-kos">{</span><span class="pl-s1">e</span><span class="pl-kos">}</span></span>'</span>)
<span class="pl-s1">airflow</span>.<span class="pl-s1">exceptions</span>.<span class="pl-v">SerializationError</span>: <span class="pl-v">Failed</span> <span class="pl-s1">to</span> <span class="pl-s1">serialize</span> <span class="pl-v">DAG</span> <span class="pl-s">'dynamic_dag_test'</span>: <span class="pl-s1">list</span> <span class="pl-s1">indices</span> <span class="pl-s1">must</span> <span class="pl-s1">be</span> <span class="pl-s1">integers</span> <span class="pl-c1">or</span> <span class="pl-s1">slices</span>, <span class="pl-c1">not</span> <span class="pl-v">Encoding</span></pre></div>
<h3 dir="auto">What you think should happen instead</h3>
<p dir="auto">Dag should contain 1 task <code class="notranslate">logging_with_operator_task</code> that contains 3 indices</p>
<h3 dir="auto">How to reproduce</h3>
<p dir="auto">copy/paste dag code into a dag file and run on airflow 2.3.0. Airflow UI will flag the error</p>
<h3 dir="auto">Operating System</h3>
<p dir="auto">Debian</p>
<h3 dir="auto">Versions of Apache Airflow Providers</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Deployment</h3>
<p dir="auto">Astronomer</p>
<h3 dir="auto">Deployment details</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Anything else</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Are you willing to submit PR?</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox"> Yes I am willing to submit a PR!</li>
</ul>
<h3 dir="auto">Code of Conduct</h3>
<ul class="contains-task-list">
<li class="task-list-item"><input type="checkbox" id="" disabled="" class="task-list-item-checkbox" checked=""> I agree to follow this project's <a href="https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md">Code of Conduct</a></li>
</ul>