MSGraphAsyncOperator fails to call API

Apache Airflow Provider(s)

microsoft-azure

Versions of Apache Airflow Providers

apache-airflow-providers-microsoft-azure==12.6.0

Apache Airflow version

3.0.4

Operating System

Debian GNU/Linux 12 (bookworm)

Deployment

Official Apache Airflow Helm Chart

Deployment details

Deployed in AKS using an extended docker image containing authorization logic and extra provider packages.
The KubernetesExecutor is used for running Tasks.

What happened

The MsGraphAsync operator fails to get an connection. It errors out with the following exception.

[2025-08-11, 09:33:08] ERROR - An error occurred: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.: source="airflow.providers.microsoft.azure.triggers.msgraph.MSGraphTrigger"
RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/triggers/msgraph.py", line 197 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 403 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 422 in send_request
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 252 in get_conn
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/hooks/base.py", line 64 in get_connection
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/connection.py", line 478 in get_connection_from_secrets
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 144 in get
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/context.py", line 155 in _get_connection
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 713 in send
File "/home/airflow/.local/lib/python3.12/site-packages/asgiref/sync.py", line 186 in __call__

The dag code:

from datetime import datetime, timedelta

from airflow import DAG
from airflow.providers.microsoft.azure.operators.msgraph import MSGraphAsyncOperator

default_args = {
    "start_date": datetime(2025, 1, 1),
    "retries": 0,
    "retry_delay": timedelta(minutes=5),
}

subscription_id = "placeholder"
resource_group = "rg-fabric"
capacity = "placeholder"

with DAG(
    dag_id="get_fabric_capacity",
    schedule=None,
    default_args=default_args,
    catchup=False,
    tags=["Fabric"]
) as dag:
    get_capacity_details = MSGraphAsyncOperator(
        task_id="get_fabric_capacity",
        conn_id="FABRIC_CAPACITY",
        method="GET",
        url=(
            "https://management.azure.com"
            f"/subscriptions/{subscription_id}"
            f"/resourceGroups/{resource_group}"
            f"/providers/Microsoft.Fabric"
            f"/capacities/{capacity}"
        ),
        query_parameters={"api-version": "2023-11-01"},
        response_type="str"
    )

What you think should happen instead

The expected behaviour is that the request is made to the API, and a response is shown, either 401 or 200 containing a response body

How to reproduce

Create a dag calling the MSGraphAsyncOperator with an existing connection and run the Dag.

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

hm this requires async connection implemention

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

This is also related to #53447
It seems that all the async operations that use the basehook for getting a connection are broken.

not all only connection s that are using sync in async session

@dabla

cc @dabla

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

The MSGraphOperator is purely async, all operaions are done through a triggerer, which is async, there are no sync operations involved (except the next method).

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

The MSGraphOperator is purely async, all operaions are done through a triggerer, which is async, there are no sync operations involved (except the next method).

no the get connection called from sync i believe it should be async

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

The MSGraphOperator is purely async, all operaions are done through a triggerer, which is async, there are no sync operations involved (except the next method).

no the get connection called from sync i believe it should be async

That must something new (or something has changed in Airflow) as this has always been the case

I will test this tomorrow

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

The MSGraphOperator is purely async, all operaions are done through a triggerer, which is async, there are no sync operations involved (except the next method).

no the get connection called from sync i believe it should be async

That must something new (or something has changed in Airflow) as this has always been the case

I think from day one the triggers must implement everything in async way, before AF3 its allowable sync version function to write and execute, but with AF3 the interaction to task sdk from the trigger should be async always otherwise the same error as seen above will happen :)

Alteast sync function should use sync_to_async.

looks like https://github.com/apache/airflow/blob/main/providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/msgraph.py#L252 this is being used for both sync and async

The MSGraphOperator is purely async, all operaions are done through a triggerer, which is async, there are no sync operations involved (except the next method).

no the get connection called from sync i believe it should be async

That must something new (or something has changed in Airflow) as this has always been the case

I think from day one the triggers must implement everything in async way, before AF3 its allowable sync version function to write and execute, but with AF3 the interaction to task sdk from the trigger should be async always otherwise the same error as seen above will happen :)

Alteast sync function should use sync_to_async.

Weird, cause we are using the MSGraphAsyncOperator in Airflow 3 and never had this issue, I even started a new PR to allow using it through start_from_trigger directly and even there we didn’t encounter this issue. Will check it out tomorrow with above release cause we are using the newer version (which is not release yet due to another bug in the triggerer for which I also did a PR).

Please note that the issue is in provider connection, previously there were issues in triggerer deadlock that being fixed. so the real fix now require in provider connection retrieval. if anyone of you willing to contribute please let me know i will assign this to you :)
I dont have bandwidth to work until nextweek.

So, would that mean an asynchronous implementation of get_connection is needed, or am I misunderstanding? I’m a bit surprised that this change was introduced without apparent testing, as it seems like it could have a significant impact.

it require async get_connection implementation in provider. if you looks at aws and google providers they all have async implementation for getting connection in triggerer. so similar way azure should have async implementation and thats the correct way, previously it was worked as its direct db call and while calling the get_connection the event loop was in block state , this is not good :)

It just that now async means async now :) so at-least sync_to_async functions should be used if anyone dont want to implement separate async functions

I would call it as provider issue this is not from airflow.

Please note that the issue is in provider connection, previously there were issues in triggerer deadlock that being fixed. so the real fix now require in provider connection retrieval. if anyone of you willing to contribute please let me know i will assign this to you :)
I dont have bandwidth to work until nextweek.

So, would that mean an asynchronous implementation of get_connection is needed, or am I misunderstanding? I’m a bit surprised that this change was introduced without apparent testing, as it seems like it could have a significant impact.

it require async get_connection implementation in provider. if you looks at aws and google providers they all have async implementation for getting connection in triggerer. so similar way azure should have async implementation and thats the correct way, previously it was worked as its direct db call and while calling the get_connection the event loop was in block state , this is not good :)

It just that now async means async now :) so at-least sync_to_async functions should be used if anyone dont want to implement separate async functions

I would call it as provider issue this is not from airflow.

Ok, I understand now, but that would imply a lot of providers will need to be updated, as those will suffer the same issue.

Yeah have todo list to find what all providers using sync call from triggerer and list them and fix with help of community , i need get list. will get it this week

I would like to open a PR to fix this, but I'm not sure if I know the correct way to solve this.

What i figure is that a quick and dirty fix would be to fetch the connection using a sync-to-async call. The propper fix would be to make sure the MsGraphAsyncOperator is not calling any synchronous methods.

Is this assumption correct?

So yes, your assumption was correct. I'm looking into it as the same issue will also arise with PowerBI.

@gopidesupavan

@gopidesupavan @gschuurman I've created a PR which fixes this issue.

@dabla

@dabla I just checked the fix, but unfortunatly it does not seem to work.

An error occurred: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.: source="airflow.providers.microsoft.azure.triggers.msgraph.MSGraphTrigger"
RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/triggers/msgraph.py", line 196 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 437 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 455 in send_request
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 354 in get_async_conn
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 253 in _build_request_adapter
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 162 in extra_dejson
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/secrets_masker.py", line 134 in mask_secret
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 740 in send
File "/home/airflow/.local/lib/python3.12/site-packages/asgiref/sync.py", line 186 in __call__

@dabla I just checked the fix, but unfortunatly it does not seem to work.

An error occurred: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.: source="airflow.providers.microsoft.azure.triggers.msgraph.MSGraphTrigger"
RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/triggers/msgraph.py", line 196 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 437 in run
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 455 in send_request
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 354 in get_async_conn
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/providers/microsoft/azure/hooks/msgraph.py", line 253 in _build_request_adapter
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 162 in extra_dejson
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/sdk/execution_time/secrets_masker.py", line 134 in mask_secret
File "/home/airflow/.local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 740 in send
File "/home/airflow/.local/lib/python3.12/site-packages/asgiref/sync.py", line 186 in __call__

Indeed, I'm also encountering the same error in 3.0.6, it worked in 3.0.4 and 3.0.5-rc2.

I just saw a new async get_connection has been added to BaseHook in Airflow 3.0.6, will use that one instead. Now I wonder how I can make my provider cross airflow compatible?

The MSGraph is doing the same as the AwsGenericHook, so I wonder if that one would also suffer the same issue? I'm 100% confident that it worked in 3.0.5-rc2, as I just started testing with 3.0.6 today, so until today, it was working fine.

I've patched code as follow in KiotaRequestAdapterHook:

async def get_async_connection(self, conn_id: str) -> Connection:
    if hasattr(BaseHook, "aget_connection"):
        print("BaseHook has aget_connection!")
        return await BaseHook.aget_connection(conn_id=conn_id)

    print("BaseHook has no aget_connection!")
    return await sync_to_async(BaseHook.get_connection)(conn_id=conn_id)


async def get_async_conn(self) -> RequestAdapter:
    print("Patched get_async_conn")
    """Initiate a new RequestAdapter connection asynchronously."""
    if not self.conn_id:
        raise AirflowException("Failed to create the KiotaRequestAdapterHook. No conn_id provided!")

    connection = await self.get_async_connection(conn_id=self.conn_id)
    api_version, request_adapter = self._build_request_adapter(connection)
    self.api_version = api_version
    return request_adapter

I'm still getting same error, here is the stacktrace:

025-09-02 11:07:27 [info     ] Traceback (most recent call last): [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/airflow/plugins/infrabel/operators/msgraph.py", line 85, in run [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     response = await self.hook.run( [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]                ^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/providers/microsoft/msgraph/hooks/msgraph.py", line 437, in run [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     response = await self.send_request( [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]                ^^^^^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/providers/microsoft/msgraph/hooks/msgraph.py", line 455, in send_request [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     conn = await self.get_async_conn() [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]            ^^^^^^^^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/airflow/plugins/infrabel/operators/msgraph.py", line 162, in get_async_conn [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     api_version, request_adapter = self._build_request_adapter(connection) [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]                                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/providers/microsoft/msgraph/hooks/msgraph.py", line 253, in _build_request_adapter [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     config = connection.extra_dejson if connection.extra else {} [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]              ^^^^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/sdk/definitions/connection.py", line 162, in extra_dejson [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     mask_secret(extra)         [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/sdk/execution_time/secrets_masker.py", line 134, in mask_secret [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     comms.send(MaskSecret(value=secret, name=name)) [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/airflow/jobs/triggerer_job_runner.py", line 740, in send [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     return async_to_sync(self.asend)(msg) [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]   File "/usr/local/lib/python3.12/site-packages/asgiref/sync.py", line 186, in __call__ [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]     raise RuntimeError(        [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ] RuntimeError: You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly. [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:27 [info     ]                                [airflow.jobs.triggerer_job_runner] chan=stdout
2025-09-02 11:07:28 [info     ] Handle event for trigger 32    [airflow.jobs.triggerer_job_runner]
2025-09-02 11:07:28 [info     ] Event submitted for trigger 32: {'status': 'failure', 'message': "You cannot use AsyncToSync in the same thread as an async event loop - just await the async function directly.

It seems the connection.extra_dejson is causing this issue:

@property
def extra_dejson(self) -> dict:
    """Returns the extra property by deserializing json."""
    from airflow.sdk.log import mask_secret

    extra = {}
    if self.extra:
        try:
            extra = json.loads(self.extra)
        except JSONDecodeError:
            log.exception("Failed to deserialize extra property `extra`, returning empty dictionary")
        else:
            mask_secret(extra)

    return extra

Probably the mask_secret is causing the RuntimeError.

I patched the _build_request_adapter method of the KiotaRequestAdapterHook locally and replace following line:

config = connection.extra_dejson if connection.extra else {}

With:

config = json.loads(connection.extra) if connection.extra else {}

And now it works again locally in 3.0.6.

I've created a new PR which avoids above mentioned issue.

Use extra_dejson in the msgraph sync get_conn now apache#54350 is fixed

Navigation Menu

Repository navigation

MSGraphAsyncOperator fails to call API #54350

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Apache Airflow version

Operating System

Deployment

Deployment details

What happened

What you think should happen instead

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

13 remaining items

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

MSGraphAsyncOperator fails to call API #54350

Description

Apache Airflow Provider(s)

Versions of Apache Airflow Providers

Apache Airflow version

Operating System

Deployment

Deployment details

What happened

What you think should happen instead

How to reproduce

Anything else

Are you willing to submit PR?

Code of Conduct

Activity

gopidesupavan commented on Aug 11, 2025

gopidesupavan commented on Aug 11, 2025

gschuurman commented on Aug 11, 2025

gopidesupavan commented on Aug 11, 2025

eladkal commented on Aug 11, 2025

dabla commented on Aug 11, 2025

gopidesupavan commented on Aug 11, 2025

dabla commented on Aug 11, 2025

dabla commented on Aug 11, 2025

gopidesupavan commented on Aug 11, 2025

dabla commented on Aug 11, 2025

13 remaining items

gopidesupavan commented on Aug 14, 2025

dabla commented on Aug 14, 2025

gopidesupavan commented on Aug 14, 2025

dabla commented on Aug 14, 2025

dabla commented on Aug 18, 2025

gschuurman commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

dabla commented on Sep 2, 2025

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions