Airflow AttributeError: 'coroutine' object has no attribute 'update_relative'

I have my dag as such:

from datetime import timedelta
import pendulum
from airflow.decorators import dag
from stage import stage_data
from table_async_pg import main

@dag(
dag_id = "data-sync",
schedule_interval = "1 * * * *",
start_date=pendulum.datetime(2023, 3, 8, tz="Asia/Hong_Kong"),
catchup=False,
dagrun_timeout=timedelta(minutes=20),
)
def Pipeline():
    a = stage_data()
    b = main()

    a >> b

pipeline = Pipeline()

When I don't include b=main() and a>>b everything works fine. If I include then I get this error:

/opt/airflow/dags/airflow_sched.py | Traceback (most recent call last):                                                                      
                                   |   File "/opt/.venv/lib/python3.9/site-packages/airflow/models/taskmixin.py", line 230, in set_downstream
                                   |     self._set_relatives(task_or_task_list, upstream=False, edge_modifier=edge_modifier)                 
                                   |   File "/opt/.venv/lib/python3.9/site-packages/airflow/models/taskmixin.py", line 175, in _set_relatives
                                   |     task_object.update_relative(self, not upstream)                                                     
                                   | AttributeError: 'coroutine' object has no attribute 'update_relative'         

the main() function you're calling in your DAG is a asyncio coroutine, which doesn't have the update_relative attribute that Airflow expects when setting task dependencies using the >> operator.

for fix this problem:

modify the main() function to return a non-coroutine function or use the >> operator with a non-coroutine function instead

def Pipeline():
    async def run_main():
        await main()

    a = stage_data()
    b = PythonOperator(
        task_id="run_main",
        python_callable=run_main,
    )

    a >> b

pipeline = Pipeline()

like this How to provide an async function in PythonOperator's python_callable in Airflow?

You're using table_async_pg.main as if it's an Airflow task, but it's not a subclass of Airflow's BaseOperator.

While you can assign any value to b, this will fail once you try to register b as an Airflow task by setting dependencies (a >> b). Airflow will try to register the dependency by calling the method update_relative, which doesn't exist on stage_data and/or main because they don't subclass BaseOperator, resulting in an error.