Add support for async callables in PythonOperator

[dabla]

This PR is related to the discussion I started on the devlist and which allows you to natively execute async code on PythonOperators.

There is also an AIP for this: https://cwiki.apache.org/confluence/display/AIRFLOW/%5BWIP%5D+AIP-98%3A+Rethinking+deferrable+operators%2C+async+hooks+and+performance+in+Airflow+3

Below an example which show you how it can be used with async hooks:

@task(show_return_value_in_logs=False)
async def load_xml_files(files):
    import asyncio
    from io import BytesIO
    from more_itertools import chunked
    from os import cpu_count
    from tenacity import retry, stop_after_attempt, wait_fixed

    from airflow.providers.sftp.hooks.sftp import SFTPClientPool

    print("number of files:", len(files))

    async with SFTPClientPool(sftp_conn_id=sftp_conn, pool_size=cpu_count()) as pool:
        @retry(stop=stop_after_attempt(3), wait=wait_fixed(5))
        async def download_file(file):
            async with pool.get_sftp_client() as sftp:
                print("downloading:", file)
                buffer = BytesIO()
                async with sftp.open(file, encoding=xml_encoding) as remote_file:
                    data = await remote_file.read()
                    buffer.write(data.encode(xml_encoding))
                    buffer.seek(0)
                return buffer

        for batch in chunked(files, cpu_count() * 2):
            tasks = [asyncio.create_task(download_file(f)) for f in batch]

            # Wait for this batch to finish before starting the next
            for task in asyncio.as_completed(tasks):
                result = await task
		 # Do something with result or accumulate it and return it as an XCom

This PR will fix additional remarks made by @kaxil on the original PR which has been reverted.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[kaxil]

Almost there, once newsfragment is added we are good

[dabla]

Almost there, once newsfragment is added we are good

And the newfragment ;-)

[dabla]

@kaxil If newsfragment is ok for you then I think it's finally ready to merge.

[kaxil]

Static check is failing, worth looking at https://github.com/apache/airflow/blob/main/airflow-core/newsfragments/54505.significant.rst or other examples

[dabla]

Since rebase of PR branch I now get following error, which is strange as the 'use next version' is defined in toml file:

common.compat changed with providers: ['standard']
common.compat provider changed but the following providers don't have '# use next version' comment for their common-compat dependency!

  - standard (providers/standard/pyproject.toml)

When common.compat changes with other providers in the same PR, add '# use next version' comment where they depend on common-compat.
Example: "apache-airflow-providers-common-compat>=1.8.0",  # use next version

To bypass this check, add the label: 'skip common compat check'

Error: Process completed with exit code 1.

[kaxil]

@dabla I think that was because the versions were bumped for the release yesterday: #60437

So you'd need to update the following line:

airflow/providers/standard/pyproject.toml

Line 77 in 33cc0ac

"apache-airflow-providers-common-compat",

to add # use next version

[dabla]

@dabla I think that was because the versions were bumped for the release yesterday: #60437

So you'd need to update the following line:

airflow/providers/standard/pyproject.toml

Line 77 in 33cc0ac

"apache-airflow-providers-common-compat",

to add # use next version

Yes indeed, cause I knew I added it, but due to rebase it was gone again as versions where indeed bumped. Re-added it now so hopefully we will be fine.

[dabla]

@kaxil last build was fine, I suppose we can merge now?

[kacpermuda]

Is this part correct @dabla ? It's breaking my AF2 dags, when installing standard provider from main. Should it have another if/elif for AF2?

Example dag:

from datetime import datetime

from airflow.providers.common.compat.assets import Asset
from airflow.providers.standard.operators.python import PythonOperator

from airflow import DAG


def do_nothing():
    pass

DAG_ID = "test_dag"

with DAG(
    dag_id=DAG_ID,
    start_date=datetime(2021, 1, 1),
    schedule=None,
    catchup=False,
    default_args={"retries": 0},
) as dag:

    task = PythonOperator(
        task_id="task",
        python_callable=do_nothing,
        inlets=[Asset(uri="s3://bucket2/dir2/file2.txt"), Asset(uri="s3://bucket2/dir2/file3.txt")],
    )

Error:

[2026-01-16, 14:20:00 UTC] {__init__.py:77} DEBUG - Lineage called with inlets: [Dataset(uri='s3://bucket2/dir2/file2.txt', extra=None), Dataset(uri='s3://bucket2/dir2/file3.txt', extra=None)], outlets: []
[2026-01-16, 14:20:00 UTC] {taskinstance.py:3336} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 282, in _run_raw_task
    TaskInstance._execute_task_with_callbacks(
  File "/usr/local/lib/python3.12/site-packages/airflow/models/taskinstance.py", line 3194, in _execute_task_with_callbacks
    self.task.post_execute(context=context, result=result)
  File "/usr/local/lib/python3.12/site-packages/airflow/lineage/__init__.py", line 88, in wrapper
    self.xcom_push(context, key=PIPELINE_INLETS, value=inlets)
TypeError: 'bool' object is not callable

Full logs:

dag_id=test_dag_run_id=manual__2026-01-16T14_19_59.788247+00_00_task_id=task_attempt=1.log

[dabla]

Hello @kacpermuda, thanks for reporting this is indeed an issue will check asap.

[dabla]

Open PR for this.