Add Azure Analysis Services model refresh support

[aaron-y-chen]

closes: #51377

Summary

• Add an azure_analysis_services connection type using Microsoft Entra service-principal client-secret authentication.
• Add a Hook, Operator, Sensor, and Trigger for starting and monitoring model refreshes in synchronous, fire-and-forget, and deferrable modes.
• Add request and polling timeouts, response validation, diagnostic HTTP errors, provider metadata, documentation, and a system-test Dag.

Testing

Run the system-test Dag in Breeze against a live Azure Analysis Services instance:

• Azure Analysis Services D1 Developer tier in West US
• Compatibility-level-1200 model
• Dedicated Microsoft Entra service principal and client-secret Airflow connection
• Two calculate refreshes

The Dag runs a fire-and-forget Operator, a deferrable-enabled Sensor, and a deferrable-enabled Operator. All three task instances complete successfully on their first try.

Verify both refresh IDs returned by Airflow through the Azure Analysis Services REST API:

Refresh ID	Type	Status	Duration
437aa86c-…-8169e	calculate	succeeded	0.206 s
d9f4974e-…-9020	calculate	succeeded	0.179 s

The Azure screenshot confirms the D1 resource, West US location, and compatibility-level-1200 model.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [GPT 5.6-sol] following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[dabla]

The PR adds a well-structured set of components (Hook, Operator, Sensor, Trigger) for Azure Analysis Services model refreshes. The architecture is sound: the trigger correctly offloads synchronous HTTP calls to asyncio.to_thread, deferrable mode is properly implemented, and the test suite is thorough.

Two items must be fixed before merge:

1. _get_base_url allows userinfo and port injection (CWE-918). The host validation explicitly guards against path/query/fragment but silently accepts user@host (userinfo) and host:port inputs, which can redirect the AAS bearer token to an attacker-controlled server. Adding or parsed_host.userinfo or parsed_host.port to the guard closes the gap. A test case should also be added.

2. requests must not be added as a dependency. httpx is already provided by the provider's transitive dependency on kiota-http and is the established HTTP client in this provider. The two requests.get / requests.post calls are straightforward to replace with the httpx sync API — see the inline comment for the exact diff.

---

Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting

[dabla]

[blocker] requests must not be added as a new dependency -- httpx is already a transitive dependency of the provider via kiota-http and is the established HTTP client across the provider.

Beyond the dependency concern, switching to httpx.AsyncClient eliminates the asyncio.to_thread workaround in the trigger entirely: httpx has native async support, so get_refresh_status and trigger_refresh can be made proper async methods on the hook, and the trigger can await them directly.

pyproject.toml -- remove the added line:

-    "requests>=2.32.0,<3",

hooks/analysis_services.py -- swap to httpx.AsyncClient and make the methods async:

-import requests
+import httpx

-    def get_refresh_status(self, server_name: str, database: str, refresh_id: str) -> dict:
+    async def get_refresh_status(self, server_name: str, database: str, refresh_id: str) -> dict:
         try:
-            response = requests.get(refresh_url, headers=self._get_headers(), timeout=self.request_timeout)
+            async with httpx.AsyncClient() as client:
+                response = await client.get(refresh_url, headers=self._get_headers(), timeout=self.request_timeout)
             response.raise_for_status()
-        except requests.RequestException as error:
+        except httpx.HTTPError as error:
             raise AzureAnalysisServicesRefreshError(...) from error

-    def trigger_refresh(self, server_name: str, database: str, refresh_type: str) -> dict:
+    async def trigger_refresh(self, server_name: str, database: str, refresh_type: str) -> dict:
         try:
-            response = requests.post(url, json={"Type": refresh_type}, headers=self._get_headers(), timeout=self.request_timeout)
+            async with httpx.AsyncClient() as client:
+                response = await client.post(url, json={"Type": refresh_type}, headers=self._get_headers(), timeout=self.request_timeout)
             response.raise_for_status()
-        except requests.RequestException as error:
+        except httpx.HTTPError as error:
             raise AzureAnalysisServicesRefreshError(...) from error

triggers/analysis_services.py -- drop asyncio.to_thread and await directly:

-import asyncio
 ...
-            result = await asyncio.to_thread(self.hook.get_refresh_status, ...)
+            result = await self.hook.get_refresh_status(...)

The response.raise_for_status(), response.json(), response.headers, and response.text APIs are identical in httpx. Test mocks will need updating accordingly (mock.AsyncMock(spec=httpx.Response)).

---

Drafted-by: Claude Sonnet 4.6 (claude-sonnet-4.6); reviewed by @dabla before posting

[aaron-y-chen]

Oh, if httpx can be used instead of requests, then there's indeed no need to add a new dependency. Let me try to refactor it. 👀

[aaron-y-chen]

Refactor done, and passed the e2e test:

I kept asyncio.to_thread() because get_refresh_status() and ClientSecretCredential.get_token() are synchronous and are also used by the Operator, Sensor, and Hook polling paths. Offloading the call keeps the triggerer event loop non-blocking without duplicating the Hook API, and follows the existing pattern in triggers/batch.py.

[dabla]

Maybe you're not aware but you have also an async variant of ClientSecretCredential, but if we would go that route that would mean we probably need a dedicated async hook, so would leave it with asyncio.to_thread as you mentioned. Could be an idea for a follow up PR in the future.

[dabla]

@aaron-y-chen I'm just wondering if the operator couldn't be async/deferred only like the MSGraphAsyncOperator? That would mean the hook could be purely written as a native hook, polling and blocking a worker isn't the best approach anyway, hence why the MSGraphAsyncOperator is purely deferred as well. If so, that would mean the code would become even more simpler, especially on the operator side and the hook would be purely async and no more blocking polling in the worker but forcing usage of trigger mechanism, which is the purpose of it, triggers are great at polling without blocking the workers. WDYT?

Thank you for the requested changes, beside above remark, the PR is looking good an clean to me, but before merging I wanted to address above point and see what your opinion is about it?

[aaron-y-chen]

Wow, this is my first time learning that there's a purely async operator in Airflow. After doing some survey, I think AAS refresh is a good fit for this approach since it avoids occupying a worker slot while polling. It's a great idea to refactor it to be async-only now that I see the benefits. Please give me some time to implement and verify it :)

[dabla]

It's a choice, for MSGraph it was an obvious choice as the Python SDK was async only, so it made sense. Here we can choose, hence why I brought it up. Still I think polling should always be done in deferred mode and thus via a trigger to avoid blocking the worker. As this is a new hook/operator, I think it's an opportunity to make that deliberate choice as well.

[aaron-y-chen]

I see. After thinking it over, I still believe we can go with the async-only approach since it's a good fit for the AAS refresh scenario.