Feature/aip 94 airflowctl tasks operations#66318
Conversation
- Add TasksOperations.list (GET /dags/{dag_id}/tasks)
- Add TaskInstancesOperations:
- list (GET /dags/{dag_id}/dagRuns/{run_id}/taskInstances)
- get (GET /dags/{dag_id}/dagRuns/{run_id}/taskInstances/{task_id})
- get_dependencies (GET .../dependencies)
- clear (POST /dags/{dag_id}/clearTaskInstances)
Operations return API responses directly to support CLI auto-generation.
Validated via REST and airflowctl; clear triggers task re-execution.
Note: CLI output formatting error for `clear` is pre-existing and out of scope.
awadhesh14
requested review from
bugraoz93,
dheerajturaga and
potiuk
as code owners
|
Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
|
justinpakzad
left a comment
justinpakzad
left a comment
There was a problem hiding this comment.
Thanks for the PR. Left a couple of comments.
| class TasksOperations(BaseOperations): | ||
| """Tasks operations.""" | ||
|
|
||
| def list(self, dag_id: str): | ||
| """List tasks for a DAG.""" | ||
| self.response = self.client.get(f"/dags/{dag_id}/tasks") | ||
| return self.response.json() | ||
|
|
||
|
|
||
| class TaskInstancesOperations(BaseOperations): |
There was a problem hiding this comment.
These appear to be duplicated. You already defined the TasksOperations() and TaskInstancesOperations() classes above (723-762).
There was a problem hiding this comment.
Thanks, removed the duplicate code.
There was a problem hiding this comment.
Any new commands should probably be added to the integration tests which live in the airflow-ctl-tests/ directory. Also, this will fail the pre-commit checks since help texts are required (can be found in the help_texts.yaml file) for any new commands added to the operations file.
| def list(self, dag_id: str): | ||
| """List tasks for a DAG.""" | ||
| self.response = self.client.get(f"/dags/{dag_id}/tasks") | ||
| return self.response.json() |
There was a problem hiding this comment.
For the list commands, I think we should follow a similar pattern to the rest of the operations where we do something like: return super().execute_list()
| self.response = self.client.get( | ||
| f"/dags/{dag_id}/dagRuns/{dag_run_id}/taskInstances/{task_id}" | ||
| ) | ||
| return self.response.json() |
There was a problem hiding this comment.
I think we should follow the same pattern as the other operations. Return validated Pydantic models instead of raw .json() and wrap each call in a try/except catching ServerResponseError.
awadhesh14
requested review from
amoghrajesh,
ashb,
gopidesupavan and
jason810496
as code owners
|
@awadhesh14 A few things need addressing before review — see our Pull Request quality criteria. Issues found:
What to do next:
No rush — take your time. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack. 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 so that our maintainers' limited time is spent where it matters most: the conversation with you. |
| def list(self, dag_id: str) -> TaskCollectionResponse | ServerResponseError: | ||
| """List tasks for a DAG.""" | ||
| try: | ||
| self.response = self.client.get(f"/dags/{dag_id}/tasks") |
There was a problem hiding this comment.
Any reason we aren't using super().execute_list here as well?
3 participants
Add airflowctl API operations for tasks and task instances.
This adds support for:
It also wires the new operations into the API client and adds unit tests using
httpx.MockTransportwith the existingmake_api_clienthelper.Closes #66173
Was generative AI tooling used to co-author this PR?
Generated-by: Codex (GPT-5) following the guidelines