如何在Airflow中提供PythonOperator的python_callable中的异步功能？

from airflow.decorators import task
from airflow.utils.decorators import apply_defaults
import asyncio

class AsyncPythonOperator(PythonOperator):
    @apply_defaults
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

    def execute(self, context):
        loop = asyncio.get_event_loop()
        loop.run_until_complete(self.async_execute(context))

    async def async_execute(self, context):
        await self.python_callable(context)

from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime

default_args = {
    'start_date': datetime(2022, 1, 1),
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

with DAG('async_dag', default_args=default_args, schedule_interval='@daily') as dag:
    async_task = AsyncPythonOperator(
        task_id='async_task',
        python_callable=my_async_function,
        provide_context=True
    )