subhramit commented on code in PR #71976:
URL: https://github.com/apache/airflow/pull/71976#discussion_r3889451229
##########
providers/influxdb/src/airflow/providers/influxdb/hooks/influxdb3.py:
##########
@@ -205,6 +222,45 @@ def query(self, query: str) -> pd.DataFrame:
return result
+ async def query_async(self, query: str) -> list[dict[str, Any]]:
+ """
+ Run a SQL query from the triggerer and return JSON-serializable
records.
+
+ ``InfluxDBClient3.query_async`` runs the blocking Arrow Flight calls
in the event
+ loop's default executor. It is a plain coroutine that resolves once
the whole result
+ stream has been read -- InfluxDB 3 has no submit-then-poll query API,
so there is
+ nothing to poll in between. Connection setup and DataFrame conversion
are offloaded
+ with ``asyncio.to_thread`` so that no step runs on the triggerer's
event loop.
+
+ :param query: SQL query string
+ :return: List of dictionaries representing query results
+ """
+ client = await asyncio.to_thread(self.get_conn)
+ if not hasattr(client, "query_async"):
+ raise InfluxDB3AsyncQueryNotAvailableError(
+ "Deferrable mode requires an InfluxDB 3 client that exposes "
+ "InfluxDBClient3.query_async(). Reinstall or upgrade the
provider "
+ "dependencies to use influxdb3-python>=0.12.0."
+ )
+
+ try:
+ import pandas as pd
+ except ImportError as e:
+ raise AirflowOptionalProviderFeatureException(
+ "pandas is required for InfluxDB 3 query results. Install it
with: "
+ "pip install 'apache-airflow-providers-influxdb[pandas]'"
+ ) from e
+
+ result = await client.query_async(query=query, language="sql",
mode="pandas")
+
+ if not isinstance(result, pd.DataFrame):
+ raise ValueError(
+ f"Query did not return a DataFrame. "
+ f"Result type:
{type(result).__module__}.{type(result).__name__}"
+ )
+
+ return await asyncio.to_thread(_convert_dataframe_to_records, result)
Review Comment:
Serialization constraint, but youre right it's in the wrong place, the
`DataFrame` cant cross a `TriggerEvent`, so I converted in the hook. It should
be the trigger's job. Moving it there.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
