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

---

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

[uranusjr]

Why do we need a separate operator class? Can this not just be a part of PythonOperator?

[dabla]

I think this one is ready for review.

[potiuk]

Very nice and clean code !

[jason810496]

#protm

[kaxil]

If a sync send() and async asend() are called concurrently (which is now possible with mixed sync/async code), could the response reads interleave? The async version you have keeps the read inside the lock, but the sync version doesn't so curious.

[dabla]

Normally it should behave correctly, unless I missed something, as the async version checks both async and sync lock, the sync version only checks the sync lock. We are now running the async PythonOperator since right after the Summit and until today haven't experienced any issues with it, I had issues in the beginning and this is how I came up with this solution as indeed it could lead to mix-ups very quickly without those locks.

[kaxil]

Should BranchPythonOperator and ShortCircuitOperator which inherit this Operator support or reject async callables?

[dabla]

The do automatically as they inherit from PythonOperator which is an async operator by default. The BaseBranchOperator operator should not be async.

[kaxil]

We should add a test for this case specifically tbh so that a regression won't make async tasks run indefinitely.

[kaxil]

@dabla Apologies for a late review, but could you take a look at them please