Lee-W commented on code in PR #69397:
URL: https://github.com/apache/airflow/pull/69397#discussion_r3643995037
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -777,9 +779,61 @@ def list(self) -> ProviderCollectionResponse |
ServerResponseError:
return super().execute_list(path="providers",
data_model=ProviderCollectionResponse)
+def _build_task_instance_path(dag_id: str, dag_run_id: str, task_id: str,
map_index: int | None) -> str:
+ """Build the task instance API path, addressing a mapped task instance
when map_index is given."""
+ path = f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}"
+ if map_index is not None and map_index >= 0:
+ path = f"{path}/{map_index}"
+ return path
+
+
class TaskInstancesOperations(BaseOperations):
"""Task instance operations."""
+ def get(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ *,
+ suppress_error_log: bool = False,
+ ) -> TaskInstanceResponse | ServerResponseError:
+ """Get a task instance for a Dag run."""
+ path = _build_task_instance_path(
+ dag_id=dag_id, dag_run_id=dag_run_id, task_id=task_id,
map_index=map_index
+ )
+ try:
+ self.response = self.client.get(
+ path,
+ extensions={"airflowctl_suppress_error_log":
suppress_error_log},
+ )
+ return
TaskInstanceResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
Review Comment:
I also notice there's lots of such usage. Let me create a PR and see how
other reacts to it
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -777,9 +779,61 @@ def list(self) -> ProviderCollectionResponse |
ServerResponseError:
return super().execute_list(path="providers",
data_model=ProviderCollectionResponse)
+def _build_task_instance_path(dag_id: str, dag_run_id: str, task_id: str,
map_index: int | None) -> str:
+ """Build the task instance API path, addressing a mapped task instance
when map_index is given."""
+ path = f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}"
+ if map_index is not None and map_index >= 0:
+ path = f"{path}/{map_index}"
+ return path
+
+
class TaskInstancesOperations(BaseOperations):
"""Task instance operations."""
+ def get(
+ self,
+ dag_id: str,
+ dag_run_id: str,
+ task_id: str,
+ map_index: int | None = None,
+ *,
+ suppress_error_log: bool = False,
+ ) -> TaskInstanceResponse | ServerResponseError:
+ """Get a task instance for a Dag run."""
+ path = _build_task_instance_path(
+ dag_id=dag_id, dag_run_id=dag_run_id, task_id=task_id,
map_index=map_index
+ )
+ try:
+ self.response = self.client.get(
+ path,
+ extensions={"airflowctl_suppress_error_log":
suppress_error_log},
+ )
+ return
TaskInstanceResponse.model_validate_json(self.response.content)
+ except ServerResponseError as e:
+ raise e
Review Comment:
```suggestion
self.response = self.client.get(
path,
extensions={"airflowctl_suppress_error_log": suppress_error_log},
)
return
TaskInstanceResponse.model_validate_json(self.response.content)
```
we don't need this try-catch. it doesn't really do anything. i guess i might
miss a few in the reviews
--
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]
