Dynamically mapped sensors throw TypeError at DAG parse time

[MatrixManAtYrService]

Apache Airflow version

main (development)

What happened

Here's a DAG:

from datetime import datetime
from airflow import DAG
from airflow.sensors.date_time import DateTimeSensor

template = "{{{{ ti.start_date + macros.timedelta(seconds={}) }}}}"

with DAG(
    dag_id="datetime_mapped",
    start_date=datetime(1970, 1, 1),
) as dag:

    @dag.task
    def get_sleeps():
        return [30, 60, 90]

    @dag.task
    def dt_templates(sleeps):
        return [template.format(s) for s in sleeps]

    templates_xcomarg = dt_templates(get_sleeps())

    DateTimeSensor.partial(task_id="sleep", mode="reschedule").apply(
        target_time=templates_xcomarg
    )

I wanted to see if it would parse, so I ran:

$ python dags/the_dag.py

And I got this error:

Traceback (most recent call last):
  File "/Users/matt/2022/02/22/dags/datetime_mapped.py", line 23, in <module>
    DateTimeSensor.partial(task_id="sleep", mode="reschedule").apply(
  File "/Users/matt/src/airflow/airflow/models/mappedoperator.py", line 203, in apply
    deps=MappedOperator.deps_for(self.operator_class),
  File "/Users/matt/src/airflow/airflow/models/mappedoperator.py", line 287, in deps_for
    return operator_class.deps | {MappedTaskIsExpanded()}
TypeError: unsupported operand type(s) for |: 'property' and 'set'
Exception ignored in: <function OperatorPartial.__del__ at 0x10ed90160>
Traceback (most recent call last):
  File "/Users/matt/src/airflow/airflow/models/mappedoperator.py", line 182, in __del__
    warnings.warn(f"{self!r} was never mapped!")
  File "/usr/local/Cellar/[email protected]/3.9.10/Frameworks/Python.framework/Versions/3.9/lib/python3.9/warnings.py", line 109, in _showwarnmsg
    sw(msg.message, msg.category, msg.filename, msg.lineno,
  File "/Users/matt/src/airflow/airflow/settings.py", line 115, in custom_show_warning
    from rich.markup import escape
  File "<frozen importlib._bootstrap>", line 1007, in _find_and_load
  File "<frozen importlib._bootstrap>", line 982, in _find_and_load_unlocked
  File "<frozen importlib._bootstrap>", line 925, in _find_spec
  File "<frozen importlib._bootstrap_external>", line 1414, in find_spec
  File "<frozen importlib._bootstrap_external>", line 1380, in _get_spec
TypeError: 'NoneType' object is not iterable

What you expected to happen

No errors. Instead a dag with three parallel sensors.

How to reproduce

Try to use the DAG shown above.

Operating System

Mac OS Bug Sur

Versions of Apache Airflow Providers

N/A

Deployment

Virtualenv installation

Deployment details

version: 2.3.0.dev0
cloned at: 8ee8f2b34

Anything else

A comment from @ashb about this

We're assuming we can call deps on the class. Which we can for everything but a sensor.

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

[uranusjr]

Not sure if #21815 would work.

[uranusjr]

BTW do we need to fix the TypeError on error reporting?