Issue 1:
Apache Airflow version

2.5.3

What happened

Pleas consider the following scenario

from typing import TYPE_CHECKING
if TYPE_CHECKING:
from my_package import MyClass

@task
def foo(a: int, b: str, c: "MyClass"):
pass

In this case, the DAG parsing will fail with a NameError from the _infer_multiple_outputs method of _TaskDecorator class in airflow/decorators/base.py.

Please note that this is a common pattern when managing multiple tasks with varying dependencies and docker containers, and my_package might not be available in some of them.

What you think should happen instead

The problem relies in the following method using typing_extensions.get_type_hints:

@multiple_outputs.default
def _infer_multiple_outputs(self):
try:
return_type = typing_extensions.get_type_hints(self.function).get("return", Any)
except TypeError: # Can't evaluate return type.
return False
ttype = getattr(return_type, "__origin__", return_type)
return ttype == dict or ttype == Dict

which will result in the following error at DAG parse time:

E NameError: name 'MyClass' is not defined

To fix this, the method could just use the __annotations__ attribute of self.function instead which does not complain about the existence of the annotated types, like so:

@multiple_outputs.default
def _infer_multiple_outputs(self):
try:
return_type = self.function.__annotations__.get("return", Any)
except TypeError: # Can't evaluate return type.
return False
ttype = getattr(return_type, "__origin__", return_type)
return ttype == dict or ttype == Dict

This should work fine since we are interested in the return type annotation anyway.

One quick fix for the programmers having this issue could simply be to add an else clause to their if TYPE_CHECKING statement, like so:

if TYPE_CHECKING:
from my_package import MyClass
else:
from typing import Any as MyClass

How to reproduce

from airflow.annotations import task
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from my_package import MyClass

@task
def foo(a: int, b: str, c: "MyClass"):
pass

Operating System

Linux Mint 21

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

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.5.1

What happened

When using the TaskFlow API, I like to generally keep a good practice of adding type annotations in the TaskFlow functions so others reading the DAG and task code have better context around inputs/outputs, keep imports solely used for typing behind typing.TYPE_CHECKING, and utilize PEP 563 for forwarding annotation evaluations. Unfortunately, when using PEP 563 and TYPE_CHECKING just TYPE_CHECKING, DAG import errors occur with a "NameError: is not defined." exception.

What you think should happen instead

Users should be free to use PEP 563 and TYPE_CHECKING when using the TaskFlow API and not hit DAG import errors along the way.

How to reproduce

Using a straightforward use case of transforming a DataFrame, let's assume this toy example:

from __future__ import annotations

from typing import TYPE_CHECKING, Any

from pendulum import datetime

from airflow.decorators import dag, task

if TYPE_CHECKING:
from pandas import DataFrame

@dag(start_date=datetime(2023, 1, 1), schedule=None)
def multiple_outputs():
@task()
def transform(df: DataFrame) -> dict[str, Any]:
...

transform()

multiple_outputs()

Add this DAG to your DAGS_FOLDER and the following import error should be observed:

Operating System

Debian GNU/Linux

Versions of Apache Airflow Providers

apache-airflow-providers-amazon==6.2.0

apache-airflow-providers-apache-hive==5.1.1

apache-airflow-providers-apache-livy==3.2.0

apache-airflow-providers-celery==3.1.0

apache-airflow-providers-cncf-kubernetes==5.1.1

apache-airflow-providers-common-sql==1.3.3

apache-airflow-providers-databricks==4.0.0

apache-airflow-providers-dbt-cloud==2.3.1

apache-airflow-providers-elasticsearch==4.3.3

apache-airflow-providers-ftp==3.3.0

apache-airflow-providers-google==8.8.0

apache-airflow-providers-http==4.1.1

apache-airflow-providers-imap==3.1.1

apache-airflow-providers-microsoft-azure==5.1.0

apache-airflow-providers-postgres==5.4.0

apache-airflow-providers-redis==3.1.0

apache-airflow-providers-sftp==4.2.1

apache-airflow-providers-snowflake==4.0.2

apache-airflow-providers-sqlite==3.3.1

apache-airflow-providers-ssh==3.4.0

astronomer-providers==1.14.0

Deployment

Astronomer

Deployment details

OOTB local Airflow install with LocalExecutor built with the Astro CLI.

Anything else

This behavior/error was not observed using Airflow 2.4.3.

As a workaround, multiple_outputs can be explicitly set on the TaskFlow function to skip the inferral.

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