Issue 1:
Apache Airflow version

2.3.0

What happened

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"Calling end {name}")

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

@task()
def show(x: dict):
print(f"x={x}")

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 >> main_run >> end_run

Could not works because of passing list of kwargs

File "/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", line 343, in _serialize
return SerializedBaseOperator.serialize_mapped_operator(var)
File "/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", 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 "/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", line 1105, in to_dict
json_dict = {"__version": cls.SERIALIZER_VERSION, "dag": cls.serialize_dag(var)}
File "/home/sang/airflow-dm-dev/venv/lib/python3.10/site-packages/airflow/serialization/serialized_objects.py", 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

What you think should happen instead

It's should accept a list of kwargs to trigger dynamic mapping instead of one kwargs

How to reproduce

No response

Operating System

Ubuntu 22.04

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.3.0 (latest released)

What happened

The following DAG was written and expected to generate 3 tasks (one for each string in the list)

dag_code

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("we've made it into the method")
logging.info(f"operator log - {string}")

@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="logging_with_operator_task",
python_callable=log_strings_operator)
.expand(op_kwargs=[{"string": "a"}, {"string": "b"}, {"string": "c"}]))

return op2

tutorial_etl_dag = tutorial_taskflow_api_etl()

error message

Broken DAG: [/usr/local/airflow/dags/dynamic_dag_test.py] Traceback (most recent call last):
File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 343, in _serialize
return SerializedBaseOperator.serialize_mapped_operator(var)
File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", 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 "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", line 1105, in to_dict
json_dict = {"__version": cls.SERIALIZER_VERSION, "dag": cls.serialize_dag(var)}
File "/usr/local/lib/python3.9/site-packages/airflow/serialization/serialized_objects.py", 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

What you think should happen instead

Dag should contain 1 task logging_with_operator_task that contains 3 indices

How to reproduce

copy/paste dag code into a dag file and run on airflow 2.3.0. Airflow UI will flag the error

Operating System

Debian

Versions of Apache Airflow Providers

No response

Deployment

Astronomer

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct