You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Suraj-kumar00 (via GitHub)" <gi...@apache.org> on 2026/03/18 16:04:45 UTC
[PR] Feat/task instance cli support [airflow]
Suraj-kumar00 opened a new pull request, #63889:
URL: https://github.com/apache/airflow/pull/63889
Adds Task Instance management support to the airflowctl CLI, enabling users to get, list, clear, and update task instances directly from the terminal.
closes: #61547
### Changes
**airflow-ctl/src/airflowctl/api/operations.py**
- Added [TaskInstanceOperations](cci:2://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:896:0-948:19) class with [get](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:371:4-377:19), [list](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:605:4-639:9), [clear](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:919:4-930:19), and [update](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:481:4-492:19) methods.
- Uses existing auto-generated Pydantic models for request/response validation.
- Intelligently unwraps Airflow's nested [TaskInstanceCollectionResponse](cci:2://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/datamodels/generated.py:2021:0-2027:63) wrappers natively for endpoints returning [get](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:371:4-377:19) and [update](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:481:4-492:19) operations to prevent Pydantic validation crashes.
**airflow-ctl/src/airflowctl/api/client.py**
- Added `task_instances` property to `Client`, enabling the auto-generated CLI commands to wire up successfully.
**airflow-ctl/src/airflowctl/ctl/cli_config.py**
- Added `"clear"` to `output_command_list` for proper CLI output rendering.
- Enhanced API field parser to automatically convert comma-separated string CLI arguments into properly formatted arrays/lists when the target Pydantic API model expects a list (e.g. `['task_1', 'task_2']` for `--task-id`).
**airflow-ctl/tests/airflow_ctl/api/test_operations.py**
- Added comprehensive [TestTaskInstanceOperations](cci:2://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/tests/airflow_ctl/api/test_operations.py:1711:0-1871:54) test suite with 6 focused unit tests.
- Covers [get](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:371:4-377:19), [list](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:605:4-639:9), [clear](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:919:4-930:19) (default and with options), and [update](cci:1://file:///Users/suraj/Documents/Personal-work/airflow/airflow-ctl/src/airflowctl/api/operations.py:481:4-492:19) (state change and note addition).
- Follows the existing XCom test pattern from #61021.
**airflow-ctl-tests/tests/airflowctl_tests/test_airflowctl_commands.py**
- Added end-to-end integration tests mapping all 4 `taskinstance` commands to their expected execution hashes.
### New CLI Commands
```bash
airflowctl taskinstance list --dag-id <dag_id> --dag-run-id <run_id>
airflowctl taskinstance get --dag-id <dag_id> --dag-run-id <run_id> --task-id <task_id>
airflowctl taskinstance clear --dag-id <dag_id> [--dry-run] [--only-failed] [--task-ids ...]
airflowctl taskinstance update --dag-id <dag_id> --dag-run-id <run_id> --task-id <task_id> [--new-state <state>] [--note <note>]
```
<img width="582" height="371" alt="Screenshot 2026-03-18 at 9 32 30 PM" src="https://github.com/user-attachments/assets/c8d8e3a4-648d-4b31-ad44-869f9fafca3a" />
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#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](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4300878079
@Suraj-kumar00 This draft PR has not had author activity for 14 days. Before anyone can review it, it needs to meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria). Here is a summary of what still needs work; once those items are addressed, mark the PR ready for review.
**Issues found:**
- :x: **Merge conflicts**: this branch has conflicts with `main` and cannot be merged as-is. Rebase: `git fetch upstream main && git rebase upstream/main`, resolve, force-push.
- :x: **Failing CI**: `Additional PROD image tests / Airflow CTL integration tests with PROD image`, `Low dep tests: providers / All-prov:LowestDeps:14:3.10:segment...zendesk`, `CI image checks / Static checks`. Please investigate and fix.
- :x: **Unresolved review comments** (14 threads (from maintainers)): please walk through each unresolved review thread. **Even if a suggestion looks incorrect or irrelevant** — and some of them will be, especially any comments left by automated reviewers like GitHub Copilot — it is still the author's responsibility to respond: apply the fix, reply in-thread with a brief explanation of why the suggestion does not apply, or resolve the thread if the feedback is no longer relevant. **Once you believe a thread is resolved — whether by pushing a fix or by explaining why the suggestion doesn't apply — please mark it as resolved yourself by clicking the 'Resolve conversation' button at the bottom of each thread.** Reviewers don't auto-close their own threads, so an addressed-but-unresolved thread reads as 'still waiting on the author' and keeps the PR from moving forward.
**What to do next:**
- Address the items above.
- Make sure static checks pass locally: `prek run --from-ref main --stage pre-commit`.
- Mark the PR as "Ready for review" when you are done.
There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "Suraj-kumar00 (via GitHub)" <gi...@apache.org>.
Suraj-kumar00 commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4172277385
Hello @potiuk, I have made the changes accordingly. Could you please review and help me with feedback also I have tried passing the CI locally but need to check here and for that as I can see it needs approval.
Let me anythings needs else from my end.
Thanks for your help!!!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 closed pull request #63889: Add task instance management commands to airflowctl CLI
URL: https://github.com/apache/airflow/pull/63889
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3069707959
##########
airflow-ctl/tests/airflow_ctl/ctl/commands/test_pool_command.py:
##########
@@ -174,8 +174,12 @@ def test_export_json_to_file(self, mock_client, tmp_path, capsys):
# Verify output message
captured = capsys.readouterr()
- expected_output = f"Exported {len(exported_data)} pool(s) to {export_file}"
- assert expected_output in captured.out.replace("\n", "")
+ import re
Review Comment:
I am not sure why this is needed 🤔
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4231883754
Still some conflicts and unresolved conversations. Converting this to draft. Feel free to reopen when you think it is ready. Thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3024424596
##########
airflow-ctl/src/airflowctl/ctl/cli_config.py:
##########
@@ -645,10 +655,21 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
datamodel_param_name = parameter_key
if expanded_parameter in self.excluded_parameters:
continue
- if expanded_parameter in args_dict.keys():
+ if (
+ expanded_parameter in args_dict.keys()
+ and args_dict[expanded_parameter] is not None
+ ):
+ val = args_dict[expanded_parameter]
Review Comment:
>Improved API field parsing to automatically convert comma-separated CLI inputs into lists when required by Pydantic models (e.g. --task-ids t1,t2 → ["t1", "t2"])
I don't quite follow your changes here. This is not the input part of the CLI. This part parses the Pydantic models from imported modules and files.
We should handle I/O in the link below. Because this is the only CLI part of auto-generated commands with client-generated. We also handle output here if you check there
https://github.com/bugraoz93/airflow/blob/ea552cb850b0e1ea66560197038d09608c7ef0b3/airflow-ctl/src/airflowctl/ctl/cli_config.py#L620-L666
##########
uv.lock:
##########
@@ -12,7 +12,7 @@ resolution-markers = [
]
[options]
-exclude-newer = "2026-03-27T13:44:39.000056016Z"
+exclude-newer = "2026-03-28T18:14:32.869345Z"
Review Comment:
Why has ' uv. lock ' been updated?
##########
airflow-ctl/src/airflowctl/api/client.py:
##########
@@ -415,6 +416,12 @@ def xcom(self):
"""Operations related to XComs."""
return XComOperations(self)
+ @lru_cache() # type: ignore[prop-decorator]
+ @property
+ def task_instances(self):
Review Comment:
Can we rename it to `tasks` to fit the current Airflow CLI to make migrations easier?
`airflowctl tasks ...`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4383014520
@Suraj-kumar00 This draft PR has been inactive for 12 days since the last triage comment and no response from the author. Closing to keep the queue clean.
You are welcome to reopen this PR when you resume work, or to open a new one addressing the issues previously raised. There is no rush — take your time.
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3025344261
##########
airflow-ctl/src/airflowctl/ctl/cli_config.py:
##########
@@ -645,10 +655,21 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
datamodel_param_name = parameter_key
if expanded_parameter in self.excluded_parameters:
continue
- if expanded_parameter in args_dict.keys():
+ if (
+ expanded_parameter in args_dict.keys()
+ and args_dict[expanded_parameter] is not None
+ ):
+ val = args_dict[expanded_parameter]
+ # Automatically convert comma-separated strings to lists if the field expects a list
+ field_annotation = str(
+ datamodel.model_fields[expanded_parameter].annotation
+ ).lower()
+ if "list" in field_annotation and isinstance(val, str):
+ val = [v.strip() for v in val.split(",") if v.strip()]
Review Comment:
`datamodel.model_fields[expanded_parameter]` can raise `KeyError` when a CLI arg exists in `args_dict` but the Pydantic model uses a different field name/alias (or the arg is for a non-model parameter). Also, detecting “list-ness” via `str(annotation).lower()` + substring `"list"` is brittle and can misclassify types. Prefer guarding with `if expanded_parameter in datamodel.model_fields`, and use `typing.get_origin()` / `typing.get_args()` (or Pydantic field metadata) to reliably detect `list[...]` and `Optional[list[...]]`.
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -893,3 +897,47 @@ def delete(
return key
except ServerResponseError as e:
raise e
+
+
+class TaskInstanceOperations(BaseOperations):
+ """Task instance operations."""
+
+ def get(self, dag_id: str, dag_run_id: str, task_id: str) -> Any:
+ """Get a task instance."""
+ self.response = self.client.get(f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}")
+ data = self.response.json()
+ if isinstance(data, list):
+ return [TaskInstanceResponse.model_validate(item) for item in data]
+ if "task_instances" in data:
+ return TaskInstanceCollectionResponse.model_validate(data)
+ return TaskInstanceResponse.model_validate(data)
Review Comment:
The method docstring says “Get a task instance”, but the implementation can return three different shapes (`list[TaskInstanceResponse]`, `TaskInstanceCollectionResponse`, or `TaskInstanceResponse`) while the signature is `-> Any`. This makes it harder for CLI/rendering code to behave predictably. Consider normalizing to a single return type for `get` (e.g., always return `TaskInstanceResponse`, unwrapping the collection/list when needed) or explicitly annotate a narrow union return type and ensure downstream formatting can handle it.
##########
airflow-ctl/tests/airflow_ctl/api/test_operations.py:
##########
@@ -1732,3 +1737,166 @@ def handle_request(request: httpx.Request) -> httpx.Response:
map_index=self.map_index,
)
assert response == self.key
+
+
+class TestTaskInstanceOperations:
+ """Test suite for Task Instance operations."""
+
+ dag_id: str = "test_dag"
+ dag_run_id: str = "manual__2025-01-24T00:00:00+00:00"
+ task_id: str = "test_task"
+
+ task_instance_response = TaskInstanceResponse(
+ id=uuid.uuid4(),
+ task_id=task_id,
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ map_index=-1,
+ logical_date=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ run_after=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ start_date=datetime.datetime(2025, 1, 24, 0, 0, 1),
+ end_date=datetime.datetime(2025, 1, 24, 0, 0, 10),
+ duration=9.0,
+ state=TaskInstanceState.SUCCESS,
+ try_number=1,
+ max_tries=0,
+ task_display_name=task_id,
+ dag_display_name=dag_id,
+ hostname="hostname",
+ unixname="airflow",
+ pool="default_pool",
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ operator="EmptyOperator",
+ executor_config="{}",
+ note=None,
+ )
+
+ task_instance_collection_response = TaskInstanceCollectionResponse(
+ task_instances=[task_instance_response],
+ total_entries=1,
+ )
+
+ def test_get(self):
+ """Test fetching a single task instance."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (
+ f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_id}"
+ )
+ return httpx.Response(200, json=json.loads(self.task_instance_response.model_dump_json()))
Review Comment:
`TaskInstanceOperations.get()` and `.update()` include branches to handle responses shaped as a raw list and as a `{"task_instances": ...}` collection wrapper, but the unit tests only cover the “single object” response path. Add tests that return (1) a JSON list payload and (2) a `TaskInstanceCollectionResponse` payload for `get`/`update` to ensure those defensive branches behave as intended.
##########
airflow-ctl/tests/airflow_ctl/api/test_operations.py:
##########
@@ -1732,3 +1737,166 @@ def handle_request(request: httpx.Request) -> httpx.Response:
map_index=self.map_index,
)
assert response == self.key
+
+
+class TestTaskInstanceOperations:
+ """Test suite for Task Instance operations."""
+
+ dag_id: str = "test_dag"
+ dag_run_id: str = "manual__2025-01-24T00:00:00+00:00"
+ task_id: str = "test_task"
+
+ task_instance_response = TaskInstanceResponse(
+ id=uuid.uuid4(),
+ task_id=task_id,
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ map_index=-1,
+ logical_date=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ run_after=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ start_date=datetime.datetime(2025, 1, 24, 0, 0, 1),
+ end_date=datetime.datetime(2025, 1, 24, 0, 0, 10),
+ duration=9.0,
+ state=TaskInstanceState.SUCCESS,
+ try_number=1,
+ max_tries=0,
+ task_display_name=task_id,
+ dag_display_name=dag_id,
+ hostname="hostname",
+ unixname="airflow",
+ pool="default_pool",
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ operator="EmptyOperator",
+ executor_config="{}",
+ note=None,
+ )
+
+ task_instance_collection_response = TaskInstanceCollectionResponse(
+ task_instances=[task_instance_response],
+ total_entries=1,
+ )
+
+ def test_get(self):
+ """Test fetching a single task instance."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (
+ f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_id}"
+ )
+ return httpx.Response(200, json=json.loads(self.task_instance_response.model_dump_json()))
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.task_instances.get(
+ dag_id=self.dag_id,
+ dag_run_id=self.dag_run_id,
+ task_id=self.task_id,
+ )
+ assert response == self.task_instance_response
+
+ def test_list(self):
+ """Test listing task instances for a DAG run."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances")
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.task_instances.list(
+ dag_id=self.dag_id,
+ dag_run_id=self.dag_run_id,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_clear(self):
+ """Test clearing task instances with default options."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == f"/api/v2/dags/{self.dag_id}/clearTaskInstances"
+ request_body = json.loads(request.content)
+ assert request_body["dry_run"] is True
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ body = ClearTaskInstancesBody(dry_run=True)
+ response = client.task_instances.clear(
+ dag_id=self.dag_id,
+ body=body,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_clear_with_options(self):
+ """Test clearing task instances with specific options."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == f"/api/v2/dags/{self.dag_id}/clearTaskInstances"
+ request_body = json.loads(request.content)
+ assert request_body["dry_run"] is False
+ assert request_body["only_failed"] is True
+ assert request_body["task_ids"] == [self.task_id]
+ assert request_body["dag_run_id"] == self.dag_run_id
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ body = ClearTaskInstancesBody(
+ dry_run=False,
+ only_failed=True,
+ task_ids=[self.task_id],
+ dag_run_id=self.dag_run_id,
+ )
+ response = client.task_instances.clear(
+ dag_id=self.dag_id,
+ body=body,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_update(self):
+ """Test updating a task instance state."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (
+ f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_id}"
+ )
+ request_body = json.loads(request.content)
+ assert request_body["new_state"] == TaskInstanceState.FAILED.value
+ return httpx.Response(200, json=json.loads(self.task_instance_response.model_dump_json()))
Review Comment:
`TaskInstanceOperations.get()` and `.update()` include branches to handle responses shaped as a raw list and as a `{"task_instances": ...}` collection wrapper, but the unit tests only cover the “single object” response path. Add tests that return (1) a JSON list payload and (2) a `TaskInstanceCollectionResponse` payload for `get`/`update` to ensure those defensive branches behave as intended.
##########
airflow-ctl/tests/airflow_ctl/ctl/commands/test_pool_command.py:
##########
@@ -174,8 +174,11 @@ def test_export_json_to_file(self, mock_client, tmp_path, capsys):
# Verify output message
captured = capsys.readouterr()
- expected_output = f"Exported {len(exported_data)} pool(s) to {export_file}"
- assert expected_output in captured.out.replace("\n", "")
+ out_str = captured.out.replace("\n", "")
+ # The output contains rich ANSI codes, so we check for key substrings instead
+ assert "Exported" in out_str
+ assert str(len(exported_data)) in out_str
+ assert "pool" in out_str
Review Comment:
These assertions became very loose (any output containing those substrings will pass), which can hide regressions in the actual message/format. Since the motivation is ANSI styling, a more robust approach is to strip ANSI escape sequences first and then assert the full expected message (or a stricter, structured subset like `f"Exported {len(exported_data)} pool(s) to {export_file}"`).
##########
airflow-ctl/src/airflowctl/api/operations.py:
##########
@@ -893,3 +897,47 @@ def delete(
return key
except ServerResponseError as e:
raise e
+
+
+class TaskInstanceOperations(BaseOperations):
+ """Task instance operations."""
+
+ def get(self, dag_id: str, dag_run_id: str, task_id: str) -> Any:
+ """Get a task instance."""
+ self.response = self.client.get(f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}")
+ data = self.response.json()
+ if isinstance(data, list):
+ return [TaskInstanceResponse.model_validate(item) for item in data]
+ if "task_instances" in data:
+ return TaskInstanceCollectionResponse.model_validate(data)
+ return TaskInstanceResponse.model_validate(data)
+
+ def list(self, dag_id: str, dag_run_id: str) -> TaskInstanceCollectionResponse | ServerResponseError:
+ """List task instances."""
+ return super().execute_list(
+ path=f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances",
+ data_model=TaskInstanceCollectionResponse,
+ )
+
+ def clear(
+ self, dag_id: str, body: ClearTaskInstancesBody
+ ) -> TaskInstanceCollectionResponse | ServerResponseError:
+ """Clear task instances."""
+ self.response = self.client.post(
+ f"dags/{dag_id}/clearTaskInstances",
+ json=body.model_dump(mode="json", exclude_unset=True),
+ )
+ return TaskInstanceCollectionResponse.model_validate_json(self.response.content)
+
+ def update(self, dag_id: str, dag_run_id: str, task_id: str, body: PatchTaskInstanceBody) -> Any:
+ """Update a task instance."""
+ self.response = self.client.patch(
+ f"dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}",
+ json=body.model_dump(mode="json", exclude_unset=True),
+ )
+ data = self.response.json()
+ if isinstance(data, list):
+ return [TaskInstanceResponse.model_validate(item) for item in data]
+ if "task_instances" in data:
+ return TaskInstanceCollectionResponse.model_validate(data)
+ return TaskInstanceResponse.model_validate(data)
Review Comment:
The response-shape normalization logic is duplicated between `get()` and `update()`. Pulling this into a private helper (e.g., `_parse_task_instance_response(data)`) would reduce repetition and make it easier to keep behavior consistent if Airflow’s API responses change.
##########
airflow-ctl/tests/airflow_ctl/api/test_operations.py:
##########
@@ -1732,3 +1737,166 @@ def handle_request(request: httpx.Request) -> httpx.Response:
map_index=self.map_index,
)
assert response == self.key
+
+
+class TestTaskInstanceOperations:
+ """Test suite for Task Instance operations."""
+
+ dag_id: str = "test_dag"
+ dag_run_id: str = "manual__2025-01-24T00:00:00+00:00"
+ task_id: str = "test_task"
+
+ task_instance_response = TaskInstanceResponse(
+ id=uuid.uuid4(),
+ task_id=task_id,
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ map_index=-1,
+ logical_date=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ run_after=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ start_date=datetime.datetime(2025, 1, 24, 0, 0, 1),
+ end_date=datetime.datetime(2025, 1, 24, 0, 0, 10),
+ duration=9.0,
+ state=TaskInstanceState.SUCCESS,
+ try_number=1,
+ max_tries=0,
+ task_display_name=task_id,
+ dag_display_name=dag_id,
+ hostname="hostname",
+ unixname="airflow",
+ pool="default_pool",
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ operator="EmptyOperator",
+ executor_config="{}",
+ note=None,
+ )
+
+ task_instance_collection_response = TaskInstanceCollectionResponse(
+ task_instances=[task_instance_response],
+ total_entries=1,
+ )
+
+ def test_get(self):
+ """Test fetching a single task instance."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (
+ f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances/{self.task_id}"
+ )
+ return httpx.Response(200, json=json.loads(self.task_instance_response.model_dump_json()))
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.task_instances.get(
+ dag_id=self.dag_id,
+ dag_run_id=self.dag_run_id,
+ task_id=self.task_id,
+ )
+ assert response == self.task_instance_response
+
+ def test_list(self):
+ """Test listing task instances for a DAG run."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == (f"/api/v2/dags/{self.dag_id}/dagRuns/{self.dag_run_id}/taskInstances")
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.task_instances.list(
+ dag_id=self.dag_id,
+ dag_run_id=self.dag_run_id,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_clear(self):
+ """Test clearing task instances with default options."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == f"/api/v2/dags/{self.dag_id}/clearTaskInstances"
+ request_body = json.loads(request.content)
+ assert request_body["dry_run"] is True
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ body = ClearTaskInstancesBody(dry_run=True)
+ response = client.task_instances.clear(
+ dag_id=self.dag_id,
+ body=body,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_clear_with_options(self):
+ """Test clearing task instances with specific options."""
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == f"/api/v2/dags/{self.dag_id}/clearTaskInstances"
+ request_body = json.loads(request.content)
+ assert request_body["dry_run"] is False
+ assert request_body["only_failed"] is True
+ assert request_body["task_ids"] == [self.task_id]
+ assert request_body["dag_run_id"] == self.dag_run_id
+ return httpx.Response(
+ 200, json=json.loads(self.task_instance_collection_response.model_dump_json())
+ )
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ body = ClearTaskInstancesBody(
+ dry_run=False,
+ only_failed=True,
+ task_ids=[self.task_id],
+ dag_run_id=self.dag_run_id,
+ )
+ response = client.task_instances.clear(
+ dag_id=self.dag_id,
+ body=body,
+ )
+ assert response == self.task_instance_collection_response
+
+ def test_update(self):
+ """Test updating a task instance state."""
Review Comment:
`TaskInstanceOperations.get()` and `.update()` include branches to handle responses shaped as a raw list and as a `{"task_instances": ...}` collection wrapper, but the unit tests only cover the “single object” response path. Add tests that return (1) a JSON list payload and (2) a `TaskInstanceCollectionResponse` payload for `get`/`update` to ensure those defensive branches behave as intended.
##########
airflow-ctl/tests/airflow_ctl/api/test_operations.py:
##########
@@ -1732,3 +1737,166 @@ def handle_request(request: httpx.Request) -> httpx.Response:
map_index=self.map_index,
)
assert response == self.key
+
+
+class TestTaskInstanceOperations:
+ """Test suite for Task Instance operations."""
+
+ dag_id: str = "test_dag"
+ dag_run_id: str = "manual__2025-01-24T00:00:00+00:00"
+ task_id: str = "test_task"
+
+ task_instance_response = TaskInstanceResponse(
+ id=uuid.uuid4(),
+ task_id=task_id,
+ dag_id=dag_id,
+ dag_run_id=dag_run_id,
+ map_index=-1,
+ logical_date=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ run_after=datetime.datetime(2025, 1, 24, 0, 0, 0),
+ start_date=datetime.datetime(2025, 1, 24, 0, 0, 1),
+ end_date=datetime.datetime(2025, 1, 24, 0, 0, 10),
+ duration=9.0,
+ state=TaskInstanceState.SUCCESS,
+ try_number=1,
+ max_tries=0,
+ task_display_name=task_id,
+ dag_display_name=dag_id,
+ hostname="hostname",
+ unixname="airflow",
+ pool="default_pool",
+ pool_slots=1,
+ queue="default",
+ priority_weight=1,
+ operator="EmptyOperator",
+ executor_config="{}",
+ note=None,
+ )
+
+ task_instance_collection_response = TaskInstanceCollectionResponse(
+ task_instances=[task_instance_response],
+ total_entries=1,
+ )
+
+ def test_get(self):
+ """Test fetching a single task instance."""
Review Comment:
`TaskInstanceOperations.get()` and `.update()` include branches to handle responses shaped as a raw list and as a `{"task_instances": ...}` collection wrapper, but the unit tests only cover the “single object” response path. Add tests that return (1) a JSON list payload and (2) a `TaskInstanceCollectionResponse` payload for `get`/`update` to ensure those defensive branches behave as intended.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "Suraj-kumar00 (via GitHub)" <gi...@apache.org>.
Suraj-kumar00 commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3047399284
##########
airflow-ctl/src/airflowctl/api/client.py:
##########
@@ -415,6 +416,12 @@ def xcom(self):
"""Operations related to XComs."""
return XComOperations(self)
+ @lru_cache() # type: ignore[prop-decorator]
+ @property
+ def task_instances(self):
Review Comment:
Done!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "Suraj-kumar00 (via GitHub)" <gi...@apache.org>.
Suraj-kumar00 commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3047407280
##########
uv.lock:
##########
@@ -12,7 +12,7 @@ resolution-markers = [
]
[options]
-exclude-newer = "2026-03-27T13:44:39.000056016Z"
+exclude-newer = "2026-03-28T18:14:32.869345Z"
Review Comment:
My local tree was unfortunately out of sync and suffering from a uv.lock divergence against upstream. I've successfully completed a clean rebase strictly onto apache:main and accepted the upstream lock file.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/task instance cli support [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4169580216
@Suraj-kumar00 This PR has a few issues that need to be addressed before it can be reviewed — please see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :warning: **Title quality**: The PR title 'Feat/task instance cli support' is a branch-name pattern rather than a properly written PR title. While it does convey the general intent, the slash-separated format ('Feat/...') is unconventional and the wording is terse. A title like 'Add task instance management commands to airflowctl CLI' would be clearer and more consistent with project norms.
- :warning: **Code quality — pointless try/except blocks**: In `operations.py`, the `get`, `clear`, and `update` methods all follow the pattern `try: ... except ServerResponseError as e: raise e`. A bare re-raise of the caught exception adds no value — it is identical to having no try/except at all. This is a common AI-generated code smell ('narrating' or boilerplate patterns) and should be removed.
**What to do next:**
- The comment informs you what you need to do.
- Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
- There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates.
- Maintainers will then proceed with a normal review.
There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #63889:
URL: https://github.com/apache/airflow/pull/63889#discussion_r3066502972
##########
airflow-ctl/src/airflowctl/ctl/cli_config.py:
##########
@@ -645,10 +656,21 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
datamodel_param_name = parameter_key
if expanded_parameter in self.excluded_parameters:
continue
- if expanded_parameter in args_dict.keys():
+ if (
+ expanded_parameter in args_dict.keys()
+ and args_dict[expanded_parameter] is not None
+ ):
+ val = args_dict[expanded_parameter]
+ if isinstance(val, str) and expanded_parameter in datamodel.model_fields:
+ import typing
+
+ annotation = datamodel.model_fields[expanded_parameter].annotation
+ origin = typing.get_origin(annotation)
+ if origin is list or getattr(annotation, "__origin__", None) is list:
+ val = [v.strip() for v in val.split(",") if v.strip()]
Review Comment:
The comma-separated-to-list conversion logic won’t trigger for common Pydantic annotations like `Optional[list[...]]` / `list[...] | None` (and often `Annotated[...]`). For example, `ClearTaskInstancesBody.task_ids` is `list[...] | None` in `api/datamodels/generated.py`, so `typing.get_origin(annotation)` will be a union, not `list`, and the value will stay a string (causing `model_validate` to fail). Consider unwrapping `Annotated`, then handling unions by checking `typing.get_args()` for a `list` origin before splitting. Also, `import typing` can be moved to module scope to avoid repeated imports inside the arg loop.
##########
airflow-ctl/src/airflowctl/ctl/cli_config.py:
##########
@@ -645,10 +656,21 @@ def _get_func(args: Namespace, api_operation: dict, api_client: Client = NEW_API
datamodel_param_name = parameter_key
if expanded_parameter in self.excluded_parameters:
continue
- if expanded_parameter in args_dict.keys():
+ if (
+ expanded_parameter in args_dict.keys()
+ and args_dict[expanded_parameter] is not None
+ ):
+ val = args_dict[expanded_parameter]
+ if isinstance(val, str) and expanded_parameter in datamodel.model_fields:
+ import typing
+
+ annotation = datamodel.model_fields[expanded_parameter].annotation
+ origin = typing.get_origin(annotation)
+ if origin is list or getattr(annotation, "__origin__", None) is list:
+ val = [v.strip() for v in val.split(",") if v.strip()]
method_params[parameter_key][
self._sanitize_method_param_key(expanded_parameter)
- ] = args_dict[expanded_parameter]
+ ] = val
Review Comment:
New behavior is introduced here (auto-splitting comma-separated CLI inputs into `list[...]` fields) but there are no unit tests covering it. Since this impacts request-body validation (and new `taskinstance clear` options), please add a CLI-config test that proves e.g. `--task-ids t1,t2` becomes `['t1','t2']` for `ClearTaskInstancesBody.task_ids`, and that non-list string fields are unaffected.
##########
airflow-ctl/src/airflowctl/api/client.py:
##########
@@ -434,6 +435,12 @@ def xcom(self):
"""Operations related to XComs."""
return XComOperations(self)
+ @lru_cache() # type: ignore[prop-decorator]
+ @property
+ def tasks(self):
Review Comment:
The new `Client` property is named `tasks`, but it actually exposes task *instance* operations (and the PR description says `task_instances`). `tasks` is ambiguous (could mean DAG tasks, not instances) and doesn’t match the naming style of other resource properties (`dag_runs`, `backfills`, etc.). Consider renaming this property to something explicit like `task_instances` (or `taskinstance` to mirror the CLI group), and update call sites/tests accordingly.
```suggestion
def task_instances(self):
```
##########
airflow-ctl/src/airflowctl/api/client.py:
##########
@@ -434,6 +435,12 @@ def xcom(self):
"""Operations related to XComs."""
return XComOperations(self)
+ @lru_cache() # type: ignore[prop-decorator]
+ @property
+ def tasks(self):
+ """Operations related to task instances."""
+ return TaskInstanceOperations(self)
Review Comment:
PR description says a `task_instances` property was added to `Client`, but the implementation adds `tasks`. Please either update the PR description or align the code to avoid confusion for reviewers and API consumers.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk closed pull request #63889: Add task instance management commands to airflowctl CLI
URL: https://github.com/apache/airflow/pull/63889
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Feat/task instance cli support [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4096402326
@Suraj-kumar00 This PR has been converted to **draft** because it does not yet meet our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **Pre-commit / static checks**: Failing: CI image checks / Static checks. Run `prek run --from-ref main` locally to find and fix issues. See [Pre-commit / static checks docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst).
- :x: **Other failing CI checks**: Failing: Additional PROD image tests / Airflow CTL integration tests with PROD image. Run `prek run --from-ref main` locally to reproduce. See [static checks docs](https://github.com/apache/airflow/blob/main/contributing-docs/08_static_code_checks.rst).
**What to do next:**
- The comment informs you what you need to do.
- Fix each issue, then mark the PR as "Ready for review" in the GitHub UI - but only after making sure that all the issues are fixed.
- There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates.
- Maintainers will then proceed with a normal review.
Converting a PR to draft is **not** a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Add task instance management commands to airflowctl CLI [airflow]
Posted by "bugraoz93 (via GitHub)" <gi...@apache.org>.
bugraoz93 commented on PR #63889:
URL: https://github.com/apache/airflow/pull/63889#issuecomment-4201310908
There are still unaddressed comments. Could you please check them?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org