dabla commented on code in PR #74600:
URL: https://github.com/apache/airflow/pull/74600#discussion_r4264776845
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/powerbi.py:
##########
@@ -42,9 +43,11 @@ class PowerBIDatasetRefreshStatus:
FAILED = "Failed"
COMPLETED = "Completed"
DISABLED = "Disabled"
+ TIMED_OUT = "TimedOut"
+ CANCELLED = "Cancelled"
- TERMINAL_STATUSES = {FAILED, COMPLETED}
- FAILURE_STATUSES = {FAILED, DISABLED}
+ FAILURE_STATUSES = {FAILED, DISABLED, TIMED_OUT, CANCELLED}
+ TERMINAL_STATUSES = FAILURE_STATUSES | {COMPLETED}
Review Comment:
**[nit]** Extending the status sets is right, but swapping their order makes
the diff look like a bigger change than it is.
`TERMINAL_STATUSES` used to come before `FAILURE_STATUSES`. The diff now
shows both lines removed and added again in the other order, so a reviewer has
to compare them side by side to see what actually changed. I see the reason:
`TERMINAL_STATUSES` is now built from `FAILURE_STATUSES`, so it has to come
second. You can keep the original order and still define one set from the other
by deriving in the opposite direction:
```python
TERMINAL_STATUSES = {FAILED, COMPLETED, DISABLED, TIMED_OUT, CANCELLED}
FAILURE_STATUSES = TERMINAL_STATUSES - {COMPLETED}
```
The diff then only touches the two set literals.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/powerbi.py:
##########
@@ -159,6 +164,9 @@ def raw_to_refresh_details(cls, refresh_details: dict) ->
dict[str, str]:
),
PowerBIDatasetRefreshFields.ERROR.value:
str(refresh_details.get("serviceExceptionJson")),
}
+ if refresh_details.get("refreshType") == "ViaEnhancedApi":
Review Comment:
**[nit]** `"ViaEnhancedApi"` is spelled out three times, and `refresh_type`
is only set for that one value.
`raw_to_refresh_details` writes the literal into the dict, and
`get_refresh_details_by_refresh_id` compares against it again at line 198. This
also changes what the public `get_refresh_history` returns, but only for
enhanced entries: some dicts have a `refresh_type` key and others don't. Two
suggestions:
- add a constant, e.g. `PowerBIDatasetRefreshType.VIA_ENHANCED_API` (or a
class attribute next to the statuses), and use it at both places;
- always store `refresh_type` (as
`str(refresh_details.get("refreshType"))`), so every entry from
`get_refresh_history` has the same keys and the check at line 198 is a plain
equality on a field that always exists.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/powerbi.py:
##########
@@ -187,6 +195,39 @@ async def get_refresh_details_by_refresh_id(
refresh_details = refresh_histories[refresh_ids.index(refresh_id)]
+ if refresh_details.get(PowerBIDatasetRefreshFields.REFRESH_TYPE.value)
== "ViaEnhancedApi":
+ try:
+ execution_details = await self.run(
+
url="myorg/groups/{group_id}/datasets/{dataset_id}/refreshes/{refresh_id}",
+ path_parameters={
+ "group_id": group_id,
+ "dataset_id": dataset_id,
+ "refresh_id": refresh_id,
+ },
+ )
+ except AirflowException as error:
+ raise PowerBIDatasetRefreshException(
+ "Failed to retrieve refresh execution details"
+ ) from error
+ status = execution_details.get("extendedStatus")
+ if status not in PowerBIDatasetRefreshStatus.TERMINAL_STATUSES |
{"NotStarted", "InProgress"}:
+ status = execution_details.get("status")
+ errors = [message["message"] for message in
execution_details.get("messages", [])]
Review Comment:
**[warning]** If a message has no `message` key, `message["message"]` raises
`KeyError`, and the trigger then cancels a refresh that is still running.
In the [execution-details
API](https://learn.microsoft.com/en-us/rest/api/power-bi/datasets/get-refresh-execution-details-in-group#enginemessage),
the `EngineMessage` fields `code`, `message` and `type` are not marked as
required. The PR also says messages are present while a refresh is still
running or retrying (`test_enhanced_refresh_messages_do_not_end_polling`). A
`KeyError` here is not a `PowerBIDatasetRefreshException`, so
`fetch_refresh_status_and_error`'s tenacity retry does not catch it. It goes
straight to `PowerBITrigger.run`'s `except Exception`, which calls
`cancel_dataset_refresh` on the live refresh. So one malformed message cancels
the user's refresh. Before this PR, nothing parsed this payload.
Reading it defensively avoids that, and is a good point to drop warnings
from the error text too. The API documents `type` as `Error` or `Warning`, and
right now a failed refresh's error message lists warnings next to the real
cause:
```python
errors = [
message["message"]
for message in execution_details.get("messages", [])
if message.get("message") and message.get("type") != "Warning"
]
```
A test with a message that has no `message` key (and one with `type:
"Warning"`) would cover both.
##########
providers/microsoft/azure/src/airflow/providers/microsoft/azure/hooks/powerbi.py:
##########
@@ -187,6 +195,39 @@ async def get_refresh_details_by_refresh_id(
refresh_details = refresh_histories[refresh_ids.index(refresh_id)]
+ if refresh_details.get(PowerBIDatasetRefreshFields.REFRESH_TYPE.value)
== "ViaEnhancedApi":
+ try:
+ execution_details = await self.run(
+
url="myorg/groups/{group_id}/datasets/{dataset_id}/refreshes/{refresh_id}",
+ path_parameters={
+ "group_id": group_id,
+ "dataset_id": dataset_id,
+ "refresh_id": refresh_id,
+ },
+ )
+ except AirflowException as error:
+ raise PowerBIDatasetRefreshException(
+ "Failed to retrieve refresh execution details"
+ ) from error
+ status = execution_details.get("extendedStatus")
+ if status not in PowerBIDatasetRefreshStatus.TERMINAL_STATUSES |
{"NotStarted", "InProgress"}:
Review Comment:
**[nit]** Enhanced refreshes report "running" as `InProgress` /
`NotStarted`, standard ones as `In Progress`.
`raw_to_refresh_details` maps only `Unknown` to
`PowerBIDatasetRefreshStatus.IN_PROGRESS`. The two active values this line
keeps from `extendedStatus` pass through unchanged. They reach the trigger's
log line and the timeout event's `dataset_refresh_status`, and the operator
pushes that value to XCom as `<task_id>.powerbi_dataset_refresh_status`. So a
downstream task reading that XCom after a timeout sees `InProgress` or
`NotStarted` for an enhanced refresh and `In Progress` for a standard one.
`test_enhanced_refresh_unknown_times_out` asserts `In Progress` only because
its fixture uses `Unknown`.
Mapping `NotStarted` and `InProgress` to `IN_PROGRESS` here, the same way
history `Unknown` is mapped, gives callers a single value for "still running".
The bare string set `{"NotStarted", "InProgress"}` could then become a named
class attribute (e.g. `ACTIVE_EXTENDED_STATUSES`) next to `TERMINAL_STATUSES`.
--
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]
