This is an automated email from the ASF dual-hosted git repository.
henry3260 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 07f5458df87 Add airflowctl dags clear command (#68706)
07f5458df87 is described below
commit 07f5458df8760e9c53e66cbc0c5b8ddf29aea044
Author: Henry Chen <[email protected]>
AuthorDate: Thu Jul 30 13:02:03 2026 +0800
Add airflowctl dags clear command (#68706)
* Add airflowctl dags clear command
* Fix airflowctl dags clear ignoring the partition date window
The Dag run list endpoint declares the partition window as
partition_date_gte / partition_date_lte. The names the command sent
belong to the request body of a different route, and FastAPI drops
query params it does not declare, so the window was silently discarded
and every run of the Dag came back — and was cleared.
partition_date_gt / partition_date_lt are not query params on that
endpoint either, so they could never have filtered anything.
* Drop unused Dag run logical date filters from airflowctl
Nothing ever passed logical_date_gt or logical_date_lt, and documenting
them broke the docs build: the spelling wordlist carries gte and lte but
not gt and lt, so the generated API docs failed spell-checking.
* Reuse the tasks clear operation for airflowctl dags clear
DagRunOperations gained a private clear helper that duplicated the
existing TasksOperations.clear. Because the CLI is generated from the
operation classes, that helper also surfaced as a subcommand named
"-clear-task-instances" with no help text.
The Dag run list filters for run ID and partition key prefix had no
callers either.
* Group airflowctl dags clear arguments with the other Dag arguments
The new arguments landed under the "# Task Commands Args" heading, so anyone
looking for what "dags clear" accepts would not find them in the Dag section
where they belong.
* Consolidate airflowctl dags clear argument validation in one place
The partition date ordering check sat in the function that fetches Dag runs
while every other check lived in the validator, so a reader had to look in
two
places to learn what the command rejects. Parsing the window up front also
removes a branch that could never be taken.
---
airflow-ctl/docs/images/command_hashes.txt | 2 +-
airflow-ctl/docs/images/output_dags.svg | 138 +++++-----
airflow-ctl/src/airflowctl/api/operations.py | 17 +-
airflow-ctl/src/airflowctl/ctl/cli_config.py | 61 +++++
.../src/airflowctl/ctl/commands/dag_command.py | 166 ++++++++++-
airflow-ctl/src/airflowctl/ctl/help_texts.yaml | 1 +
.../tests/airflow_ctl/api/test_operations.py | 37 +++
.../airflow_ctl/ctl/commands/test_dag_command.py | 305 ++++++++++++++++++++-
8 files changed, 654 insertions(+), 73 deletions(-)
diff --git a/airflow-ctl/docs/images/command_hashes.txt
b/airflow-ctl/docs/images/command_hashes.txt
index 74b5ede7ce3..40ddcd13a0b 100644
--- a/airflow-ctl/docs/images/command_hashes.txt
+++ b/airflow-ctl/docs/images/command_hashes.txt
@@ -4,7 +4,7 @@ auth:d79e9c7d00c432bdbcbc2a86e2e32053
backfill:74c8737b0a62a86ed3605fa9e6165874
config:a3d936cb15fe3b547bf6c82cf93d923f
connections:942f9f88cb908c28bf5c19159fc5065b
-dags:5ad68174a1111563dff870a241914b30
+dags:ea477b70a1c3bcf8ed603cdbd11e5ab6
dagrun:07035226eaaff0a557d3ad9aab4b41b5
jobs:a5b644c5da8889443bb40ee10b599270
pools:19efe105b9515ab1926ebcaf0e028d71
diff --git a/airflow-ctl/docs/images/output_dags.svg
b/airflow-ctl/docs/images/output_dags.svg
index 70f2cda4cd0..78d61ab4506 100644
--- a/airflow-ctl/docs/images/output_dags.svg
+++ b/airflow-ctl/docs/images/output_dags.svg
@@ -1,4 +1,4 @@
-<svg class="rich-terminal" viewBox="0 0 933 708.8"
xmlns="http://www.w3.org/2000/svg";>
+<svg class="rich-terminal" viewBox="0 0 933 757.5999999999999"
xmlns="http://www.w3.org/2000/svg";>
<!-- Generated with Rich https://www.textualize.io -->
<style>
@@ -19,146 +19,154 @@
font-weight: 700;
}
- .terminal-3245301439-matrix {
+ .terminal-4223325571-matrix {
font-family: Fira Code, monospace;
font-size: 20px;
line-height: 24.4px;
font-variant-east-asian: full-width;
}
- .terminal-3245301439-title {
+ .terminal-4223325571-title {
font-size: 18px;
font-weight: bold;
font-family: arial;
}
- .terminal-3245301439-r1 { fill: #ff8700 }
-.terminal-3245301439-r2 { fill: #c5c8c6 }
-.terminal-3245301439-r3 { fill: #808080 }
-.terminal-3245301439-r4 { fill: #68a0b3 }
+ .terminal-4223325571-r1 { fill: #ff8700 }
+.terminal-4223325571-r2 { fill: #c5c8c6 }
+.terminal-4223325571-r3 { fill: #808080 }
+.terminal-4223325571-r4 { fill: #68a0b3 }
</style>
<defs>
- <clipPath id="terminal-3245301439-clip-terminal">
- <rect x="0" y="0" width="914.0" height="657.8" />
+ <clipPath id="terminal-4223325571-clip-terminal">
+ <rect x="0" y="0" width="914.0" height="706.5999999999999" />
</clipPath>
- <clipPath id="terminal-3245301439-line-0">
+ <clipPath id="terminal-4223325571-line-0">
<rect x="0" y="1.5" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-1">
+<clipPath id="terminal-4223325571-line-1">
<rect x="0" y="25.9" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-2">
+<clipPath id="terminal-4223325571-line-2">
<rect x="0" y="50.3" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-3">
+<clipPath id="terminal-4223325571-line-3">
<rect x="0" y="74.7" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-4">
+<clipPath id="terminal-4223325571-line-4">
<rect x="0" y="99.1" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-5">
+<clipPath id="terminal-4223325571-line-5">
<rect x="0" y="123.5" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-6">
+<clipPath id="terminal-4223325571-line-6">
<rect x="0" y="147.9" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-7">
+<clipPath id="terminal-4223325571-line-7">
<rect x="0" y="172.3" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-8">
+<clipPath id="terminal-4223325571-line-8">
<rect x="0" y="196.7" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-9">
+<clipPath id="terminal-4223325571-line-9">
<rect x="0" y="221.1" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-10">
+<clipPath id="terminal-4223325571-line-10">
<rect x="0" y="245.5" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-11">
+<clipPath id="terminal-4223325571-line-11">
<rect x="0" y="269.9" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-12">
+<clipPath id="terminal-4223325571-line-12">
<rect x="0" y="294.3" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-13">
+<clipPath id="terminal-4223325571-line-13">
<rect x="0" y="318.7" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-14">
+<clipPath id="terminal-4223325571-line-14">
<rect x="0" y="343.1" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-15">
+<clipPath id="terminal-4223325571-line-15">
<rect x="0" y="367.5" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-16">
+<clipPath id="terminal-4223325571-line-16">
<rect x="0" y="391.9" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-17">
+<clipPath id="terminal-4223325571-line-17">
<rect x="0" y="416.3" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-18">
+<clipPath id="terminal-4223325571-line-18">
<rect x="0" y="440.7" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-19">
+<clipPath id="terminal-4223325571-line-19">
<rect x="0" y="465.1" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-20">
+<clipPath id="terminal-4223325571-line-20">
<rect x="0" y="489.5" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-21">
+<clipPath id="terminal-4223325571-line-21">
<rect x="0" y="513.9" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-22">
+<clipPath id="terminal-4223325571-line-22">
<rect x="0" y="538.3" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-23">
+<clipPath id="terminal-4223325571-line-23">
<rect x="0" y="562.7" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-24">
+<clipPath id="terminal-4223325571-line-24">
<rect x="0" y="587.1" width="915" height="24.65"/>
</clipPath>
-<clipPath id="terminal-3245301439-line-25">
+<clipPath id="terminal-4223325571-line-25">
<rect x="0" y="611.5" width="915" height="24.65"/>
</clipPath>
+<clipPath id="terminal-4223325571-line-26">
+ <rect x="0" y="635.9" width="915" height="24.65"/>
+ </clipPath>
+<clipPath id="terminal-4223325571-line-27">
+ <rect x="0" y="660.3" width="915" height="24.65"/>
+ </clipPath>
</defs>
- <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1"
x="1" y="1" width="931" height="706.8" rx="8"/>
+ <rect fill="#292929" stroke="rgba(255,255,255,0.35)" stroke-width="1"
x="1" y="1" width="931" height="755.6" rx="8"/>
<g transform="translate(26,22)">
<circle cx="0" cy="0" r="7" fill="#ff5f57"/>
<circle cx="22" cy="0" r="7" fill="#febc2e"/>
<circle cx="44" cy="0" r="7" fill="#28c840"/>
</g>
- <g transform="translate(9, 41)"
clip-path="url(#terminal-3245301439-clip-terminal)">
+ <g transform="translate(9, 41)"
clip-path="url(#terminal-4223325571-clip-terminal)">
- <g class="terminal-3245301439-matrix">
- <text class="terminal-3245301439-r1" x="0" y="20" textLength="73.2"
clip-path="url(#terminal-3245301439-line-0)">Usage:</text><text
class="terminal-3245301439-r3" x="85.4" y="20" textLength="183"
clip-path="url(#terminal-3245301439-line-0)">airflowctl dags</text><text
class="terminal-3245301439-r2" x="268.4" y="20" textLength="24.4"
clip-path="url(#terminal-3245301439-line-0)"> [</text><text
class="terminal-3245301439-r4" x="292.8" y="20" textLength="24.4"
clip-path="url(#t [...]
-</text><text class="terminal-3245301439-r2" x="915" y="44.4" textLength="12.2"
clip-path="url(#terminal-3245301439-line-1)">
-</text><text class="terminal-3245301439-r2" x="0" y="68.8" textLength="280.6"
clip-path="url(#terminal-3245301439-line-2)">Perform Dags operations</text><text
class="terminal-3245301439-r2" x="915" y="68.8" textLength="12.2"
clip-path="url(#terminal-3245301439-line-2)">
-</text><text class="terminal-3245301439-r2" x="915" y="93.2" textLength="12.2"
clip-path="url(#terminal-3245301439-line-3)">
-</text><text class="terminal-3245301439-r1" x="0" y="117.6" textLength="256.2"
clip-path="url(#terminal-3245301439-line-4)">Positional Arguments:</text><text
class="terminal-3245301439-r2" x="915" y="117.6" textLength="12.2"
clip-path="url(#terminal-3245301439-line-4)">
-</text><text class="terminal-3245301439-r4" x="24.4" y="142" textLength="85.4"
clip-path="url(#terminal-3245301439-line-5)">COMMAND</text><text
class="terminal-3245301439-r2" x="915" y="142" textLength="12.2"
clip-path="url(#terminal-3245301439-line-5)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="166.4"
textLength="73.2"
clip-path="url(#terminal-3245301439-line-6)">delete</text><text
class="terminal-3245301439-r2" x="268.4" y="166.4" textLength="268.4"
clip-path="url(#terminal-3245301439-line-6)">Delete a Dag by its ID</text><text
class="terminal-3245301439-r2" x="915" y="166.4" textLength="12.2"
clip-path="url(#terminal-3245301439-line-6)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="190.8"
textLength="36.6" clip-path="url(#terminal-3245301439-line-7)">get</text><text
class="terminal-3245301439-r2" x="268.4" y="190.8" textLength="292.8"
clip-path="url(#terminal-3245301439-line-7)">Retrieve a Dag by its ID</text><text
class="terminal-3245301439-r2" x="915" y="190.8" textLength="12.2"
clip-path="url(#terminal-3245301439-line-7)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="215.2"
textLength="134.2"
clip-path="url(#terminal-3245301439-line-8)">get-details</text><text
class="terminal-3245301439-r2" x="268.4" y="215.2" textLength="475.8"
clip-path="url(#terminal-3245301439-line-8)">Retrieve detailed information for a Dag</text><text
class="terminal-3245301439-r2" x="915" y="215.2" textLength="12.2"
clip-path="url(#terminal-3245301439-line-8)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="239.6"
textLength="195.2"
clip-path="url(#terminal-3245301439-line-9)">get-import-error</text><text
class="terminal-3245301439-r2" x="268.4" y="239.6" textLength="451.4"
clip-path="url(#terminal-3245301439-line-9)">Retrieve a Dag import error by its ID</text><text
class="terminal-3245301439-r2" x="915" y="239.6" textLength="12.2"
clip-path="url(#terminal-3245301439-line-9)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="264"
textLength="109.8"
clip-path="url(#terminal-3245301439-line-10)">get-stats</text><text
class="terminal-3245301439-r2" x="268.4" y="264" textLength="536.8"
clip-path="url(#terminal-3245301439-line-10)">Retrieve run statistics for one or more Dags</text><text
class="terminal-3245301439-r2" x="915" y="264" textLength="12.2"
clip-path="url(#terminal-3245301439-line-10)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="288.4"
textLength="97.6"
clip-path="url(#terminal-3245301439-line-11)">get-tags</text><text
class="terminal-3245301439-r2" x="268.4" y="288.4" textLength="366"
clip-path="url(#terminal-3245301439-line-11)">List all tags used across Dags</text><text
class="terminal-3245301439-r2" x="915" y="288.4" textLength="12.2"
clip-path="url(#terminal-3245301439-line-11)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="312.8"
textLength="134.2"
clip-path="url(#terminal-3245301439-line-12)">get-version</text><text
class="terminal-3245301439-r2" x="268.4" y="312.8" textLength="439.2"
clip-path="url(#terminal-3245301439-line-12)">Retrieve a specific version of a Dag</text><text
class="terminal-3245301439-r2" x="915" y="312.8" textLength="12.2"
clip-path="url(#terminal-3245301439-line-12)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="337.2"
textLength="48.8"
clip-path="url(#terminal-3245301439-line-13)">list</text><text
class="terminal-3245301439-r2" x="268.4" y="337.2" textLength="158.6"
clip-path="url(#terminal-3245301439-line-13)">List all Dags</text><text
class="terminal-3245301439-r2" x="915" y="337.2" textLength="12.2"
clip-path="url(#terminal-3245301439-line-13)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="361.6"
textLength="219.6"
clip-path="url(#terminal-3245301439-line-14)">list-import-errors</text><text
class="terminal-3245301439-r2" x="915" y="361.6" textLength="12.2"
clip-path="url(#terminal-3245301439-line-14)">
-</text><text class="terminal-3245301439-r2" x="268.4" y="386"
textLength="317.2"
clip-path="url(#terminal-3245301439-line-15)">List all Dag import errors</text><text
class="terminal-3245301439-r2" x="915" y="386" textLength="12.2"
clip-path="url(#terminal-3245301439-line-15)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="410.4"
textLength="146.4"
clip-path="url(#terminal-3245301439-line-16)">list-version</text><text
class="terminal-3245301439-r2" x="268.4" y="410.4" textLength="317.2"
clip-path="url(#terminal-3245301439-line-16)">List all versions of a Dag</text><text
class="terminal-3245301439-r2" x="915" y="410.4" textLength="12.2"
clip-path="url(#terminal-3245301439-line-16)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="434.8"
textLength="146.4"
clip-path="url(#terminal-3245301439-line-17)">list-warning</text><text
class="terminal-3245301439-r2" x="268.4" y="434.8" textLength="256.2"
clip-path="url(#terminal-3245301439-line-17)">List all Dag warnings</text><text
class="terminal-3245301439-r2" x="915" y="434.8" textLength="12.2"
clip-path="url(#terminal-3245301439-line-17)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="459.2"
textLength="170.8"
clip-path="url(#terminal-3245301439-line-18)">next-execution</text><text
class="terminal-3245301439-r2" x="268.4" y="459.2" textLength="585.6"
clip-path="url(#terminal-3245301439-line-18)">Show the next scheduled execution time for a Dag</text><text
class="terminal-3245301439-r2" x="915" y="459.2" textLength="12.2"
clip-path="url(#terminal-3245301439-line-18)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="483.6" textLength="61"
clip-path="url(#terminal-3245301439-line-19)">pause</text><text
class="terminal-3245301439-r2" x="268.4" y="483.6" textLength="134.2"
clip-path="url(#terminal-3245301439-line-19)">Pause a Dag</text><text
class="terminal-3245301439-r2" x="915" y="483.6" textLength="12.2"
clip-path="url(#terminal-3245301439-line-19)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="508" textLength="61"
clip-path="url(#terminal-3245301439-line-20)">state</text><text
class="terminal-3245301439-r2" x="268.4" y="508" textLength="329.4"
clip-path="url(#terminal-3245301439-line-20)">Get the status of a Dag run</text><text
class="terminal-3245301439-r2" x="915" y="508" textLength="12.2"
clip-path="url(#terminal-3245301439-line-20)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="532.4"
textLength="85.4"
clip-path="url(#terminal-3245301439-line-21)">trigger</text><text
class="terminal-3245301439-r2" x="268.4" y="532.4" textLength="256.2"
clip-path="url(#terminal-3245301439-line-21)">Trigger a new Dag run</text><text
class="terminal-3245301439-r2" x="915" y="532.4" textLength="12.2"
clip-path="url(#terminal-3245301439-line-21)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="556.8"
textLength="85.4"
clip-path="url(#terminal-3245301439-line-22)">unpause</text><text
class="terminal-3245301439-r2" x="268.4" y="556.8" textLength="158.6"
clip-path="url(#terminal-3245301439-line-22)">Unpause a Dag</text><text
class="terminal-3245301439-r2" x="915" y="556.8" textLength="12.2"
clip-path="url(#terminal-3245301439-line-22)">
-</text><text class="terminal-3245301439-r4" x="48.8" y="581.2"
textLength="73.2"
clip-path="url(#terminal-3245301439-line-23)">update</text><text
class="terminal-3245301439-r2" x="268.4" y="581.2" textLength="317.2"
clip-path="url(#terminal-3245301439-line-23)">Update properties of a Dag</text><text
class="terminal-3245301439-r2" x="915" y="581.2" textLength="12.2"
clip-path="url(#terminal-3245301439-line-23)">
-</text><text class="terminal-3245301439-r2" x="915" y="605.6"
textLength="12.2" clip-path="url(#terminal-3245301439-line-24)">
-</text><text class="terminal-3245301439-r1" x="0" y="630" textLength="97.6"
clip-path="url(#terminal-3245301439-line-25)">Options:</text><text
class="terminal-3245301439-r2" x="915" y="630" textLength="12.2"
clip-path="url(#terminal-3245301439-line-25)">
-</text><text class="terminal-3245301439-r4" x="24.4" y="654.4"
textLength="24.4" clip-path="url(#terminal-3245301439-line-26)">-h</text><text
class="terminal-3245301439-r2" x="48.8" y="654.4" textLength="24.4"
clip-path="url(#terminal-3245301439-line-26)">, </text><text
class="terminal-3245301439-r4" x="73.2" y="654.4" textLength="73.2"
clip-path="url(#terminal-3245301439-line-26)">--help</text><text
class="terminal-3245301439-r2" x="268.4" y="654.4" textLength="378.2"
clip-path="ur [...]
+ <g class="terminal-4223325571-matrix">
+ <text class="terminal-4223325571-r1" x="0" y="20" textLength="73.2"
clip-path="url(#terminal-4223325571-line-0)">Usage:</text><text
class="terminal-4223325571-r3" x="85.4" y="20" textLength="183"
clip-path="url(#terminal-4223325571-line-0)">airflowctl dags</text><text
class="terminal-4223325571-r2" x="268.4" y="20" textLength="24.4"
clip-path="url(#terminal-4223325571-line-0)"> [</text><text
class="terminal-4223325571-r4" x="292.8" y="20" textLength="24.4"
clip-path="url(#t [...]
+</text><text class="terminal-4223325571-r2" x="915" y="44.4" textLength="12.2"
clip-path="url(#terminal-4223325571-line-1)">
+</text><text class="terminal-4223325571-r2" x="0" y="68.8" textLength="280.6"
clip-path="url(#terminal-4223325571-line-2)">Perform Dags operations</text><text
class="terminal-4223325571-r2" x="915" y="68.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-2)">
+</text><text class="terminal-4223325571-r2" x="915" y="93.2" textLength="12.2"
clip-path="url(#terminal-4223325571-line-3)">
+</text><text class="terminal-4223325571-r1" x="0" y="117.6" textLength="256.2"
clip-path="url(#terminal-4223325571-line-4)">Positional Arguments:</text><text
class="terminal-4223325571-r2" x="915" y="117.6" textLength="12.2"
clip-path="url(#terminal-4223325571-line-4)">
+</text><text class="terminal-4223325571-r4" x="24.4" y="142" textLength="85.4"
clip-path="url(#terminal-4223325571-line-5)">COMMAND</text><text
class="terminal-4223325571-r2" x="915" y="142" textLength="12.2"
clip-path="url(#terminal-4223325571-line-5)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="166.4" textLength="61"
clip-path="url(#terminal-4223325571-line-6)">clear</text><text
class="terminal-4223325571-r2" x="268.4" y="166.4" textLength="646.6"
clip-path="url(#terminal-4223325571-line-6)">Clear task instances for Dag runs selected by run ID,</text><text
class="terminal-4223325571-r2" x="915" y="166.4" textLength="12.2"
clip-path="url(#terminal-4223325571-line-6)">
+</text><text class="terminal-4223325571-r2" x="0" y="190.8" textLength="390.4"
clip-path="url(#terminal-4223325571-line-7)">partition key, or partition date</text><text
class="terminal-4223325571-r2" x="915" y="190.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-7)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="215.2"
textLength="73.2"
clip-path="url(#terminal-4223325571-line-8)">delete</text><text
class="terminal-4223325571-r2" x="268.4" y="215.2" textLength="268.4"
clip-path="url(#terminal-4223325571-line-8)">Delete a Dag by its ID</text><text
class="terminal-4223325571-r2" x="915" y="215.2" textLength="12.2"
clip-path="url(#terminal-4223325571-line-8)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="239.6"
textLength="36.6" clip-path="url(#terminal-4223325571-line-9)">get</text><text
class="terminal-4223325571-r2" x="268.4" y="239.6" textLength="292.8"
clip-path="url(#terminal-4223325571-line-9)">Retrieve a Dag by its ID</text><text
class="terminal-4223325571-r2" x="915" y="239.6" textLength="12.2"
clip-path="url(#terminal-4223325571-line-9)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="264"
textLength="134.2"
clip-path="url(#terminal-4223325571-line-10)">get-details</text><text
class="terminal-4223325571-r2" x="268.4" y="264" textLength="475.8"
clip-path="url(#terminal-4223325571-line-10)">Retrieve detailed information for a Dag</text><text
class="terminal-4223325571-r2" x="915" y="264" textLength="12.2"
clip-path="url(#terminal-4223325571-line-10)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="288.4"
textLength="195.2"
clip-path="url(#terminal-4223325571-line-11)">get-import-error</text><text
class="terminal-4223325571-r2" x="268.4" y="288.4" textLength="451.4"
clip-path="url(#terminal-4223325571-line-11)">Retrieve a Dag import error by its ID</text><text
class="terminal-4223325571-r2" x="915" y="288.4" textLength="12.2"
clip-path="url(#terminal-4223325571-line-11)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="312.8"
textLength="109.8"
clip-path="url(#terminal-4223325571-line-12)">get-stats</text><text
class="terminal-4223325571-r2" x="268.4" y="312.8" textLength="536.8"
clip-path="url(#terminal-4223325571-line-12)">Retrieve run statistics for one or more Dags</text><text
class="terminal-4223325571-r2" x="915" y="312.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-12)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="337.2"
textLength="97.6"
clip-path="url(#terminal-4223325571-line-13)">get-tags</text><text
class="terminal-4223325571-r2" x="268.4" y="337.2" textLength="366"
clip-path="url(#terminal-4223325571-line-13)">List all tags used across Dags</text><text
class="terminal-4223325571-r2" x="915" y="337.2" textLength="12.2"
clip-path="url(#terminal-4223325571-line-13)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="361.6"
textLength="134.2"
clip-path="url(#terminal-4223325571-line-14)">get-version</text><text
class="terminal-4223325571-r2" x="268.4" y="361.6" textLength="439.2"
clip-path="url(#terminal-4223325571-line-14)">Retrieve a specific version of a Dag</text><text
class="terminal-4223325571-r2" x="915" y="361.6" textLength="12.2"
clip-path="url(#terminal-4223325571-line-14)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="386" textLength="48.8"
clip-path="url(#terminal-4223325571-line-15)">list</text><text
class="terminal-4223325571-r2" x="268.4" y="386" textLength="158.6"
clip-path="url(#terminal-4223325571-line-15)">List all Dags</text><text
class="terminal-4223325571-r2" x="915" y="386" textLength="12.2"
clip-path="url(#terminal-4223325571-line-15)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="410.4"
textLength="219.6"
clip-path="url(#terminal-4223325571-line-16)">list-import-errors</text><text
class="terminal-4223325571-r2" x="915" y="410.4" textLength="12.2"
clip-path="url(#terminal-4223325571-line-16)">
+</text><text class="terminal-4223325571-r2" x="268.4" y="434.8"
textLength="317.2"
clip-path="url(#terminal-4223325571-line-17)">List all Dag import errors</text><text
class="terminal-4223325571-r2" x="915" y="434.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-17)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="459.2"
textLength="146.4"
clip-path="url(#terminal-4223325571-line-18)">list-version</text><text
class="terminal-4223325571-r2" x="268.4" y="459.2" textLength="317.2"
clip-path="url(#terminal-4223325571-line-18)">List all versions of a Dag</text><text
class="terminal-4223325571-r2" x="915" y="459.2" textLength="12.2"
clip-path="url(#terminal-4223325571-line-18)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="483.6"
textLength="146.4"
clip-path="url(#terminal-4223325571-line-19)">list-warning</text><text
class="terminal-4223325571-r2" x="268.4" y="483.6" textLength="256.2"
clip-path="url(#terminal-4223325571-line-19)">List all Dag warnings</text><text
class="terminal-4223325571-r2" x="915" y="483.6" textLength="12.2"
clip-path="url(#terminal-4223325571-line-19)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="508"
textLength="170.8"
clip-path="url(#terminal-4223325571-line-20)">next-execution</text><text
class="terminal-4223325571-r2" x="268.4" y="508" textLength="585.6"
clip-path="url(#terminal-4223325571-line-20)">Show the next scheduled execution time for a Dag</text><text
class="terminal-4223325571-r2" x="915" y="508" textLength="12.2"
clip-path="url(#terminal-4223325571-line-20)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="532.4" textLength="61"
clip-path="url(#terminal-4223325571-line-21)">pause</text><text
class="terminal-4223325571-r2" x="268.4" y="532.4" textLength="134.2"
clip-path="url(#terminal-4223325571-line-21)">Pause a Dag</text><text
class="terminal-4223325571-r2" x="915" y="532.4" textLength="12.2"
clip-path="url(#terminal-4223325571-line-21)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="556.8" textLength="61"
clip-path="url(#terminal-4223325571-line-22)">state</text><text
class="terminal-4223325571-r2" x="268.4" y="556.8" textLength="329.4"
clip-path="url(#terminal-4223325571-line-22)">Get the status of a Dag run</text><text
class="terminal-4223325571-r2" x="915" y="556.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-22)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="581.2"
textLength="85.4"
clip-path="url(#terminal-4223325571-line-23)">trigger</text><text
class="terminal-4223325571-r2" x="268.4" y="581.2" textLength="256.2"
clip-path="url(#terminal-4223325571-line-23)">Trigger a new Dag run</text><text
class="terminal-4223325571-r2" x="915" y="581.2" textLength="12.2"
clip-path="url(#terminal-4223325571-line-23)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="605.6"
textLength="85.4"
clip-path="url(#terminal-4223325571-line-24)">unpause</text><text
class="terminal-4223325571-r2" x="268.4" y="605.6" textLength="158.6"
clip-path="url(#terminal-4223325571-line-24)">Unpause a Dag</text><text
class="terminal-4223325571-r2" x="915" y="605.6" textLength="12.2"
clip-path="url(#terminal-4223325571-line-24)">
+</text><text class="terminal-4223325571-r4" x="48.8" y="630" textLength="73.2"
clip-path="url(#terminal-4223325571-line-25)">update</text><text
class="terminal-4223325571-r2" x="268.4" y="630" textLength="317.2"
clip-path="url(#terminal-4223325571-line-25)">Update properties of a Dag</text><text
class="terminal-4223325571-r2" x="915" y="630" textLength="12.2"
clip-path="url(#terminal-4223325571-line-25)">
+</text><text class="terminal-4223325571-r2" x="915" y="654.4"
textLength="12.2" clip-path="url(#terminal-4223325571-line-26)">
+</text><text class="terminal-4223325571-r1" x="0" y="678.8" textLength="97.6"
clip-path="url(#terminal-4223325571-line-27)">Options:</text><text
class="terminal-4223325571-r2" x="915" y="678.8" textLength="12.2"
clip-path="url(#terminal-4223325571-line-27)">
+</text><text class="terminal-4223325571-r4" x="24.4" y="703.2"
textLength="24.4" clip-path="url(#terminal-4223325571-line-28)">-h</text><text
class="terminal-4223325571-r2" x="48.8" y="703.2" textLength="24.4"
clip-path="url(#terminal-4223325571-line-28)">, </text><text
class="terminal-4223325571-r4" x="73.2" y="703.2" textLength="73.2"
clip-path="url(#terminal-4223325571-line-28)">--help</text><text
class="terminal-4223325571-r2" x="268.4" y="703.2" textLength="378.2"
clip-path="ur [...]
</text>
</g>
</g>
diff --git a/airflow-ctl/src/airflowctl/api/operations.py
b/airflow-ctl/src/airflowctl/api/operations.py
index b2e431afd88..5a5556cbe45 100644
--- a/airflow-ctl/src/airflowctl/api/operations.py
+++ b/airflow-ctl/src/airflowctl/api/operations.py
@@ -91,7 +91,8 @@ T = TypeVar("T", bound=BaseModel)
def _serialize_query_param(value: Any) -> Any:
- if isinstance(value, datetime.datetime):
+ # datetime.datetime subclasses datetime.date, so this covers both.
+ if isinstance(value, datetime.date):
return value.isoformat()
return value
@@ -525,12 +526,16 @@ class DagRunOperations(BaseOperations):
self,
state: str | None = None,
limit: int = 100,
+ offset: int | None = None,
start_date: datetime.datetime | None = None,
end_date: datetime.datetime | None = None,
dag_id: str | None = None,
logical_date_gte: datetime.datetime | None = None,
logical_date_lte: datetime.datetime | None = None,
+ partition_date_gte: datetime.date | None = None,
+ partition_date_lte: datetime.date | None = None,
order_by: str | None = None,
+ partition_key_pattern: str | None = None,
*,
suppress_error_log: bool = False,
) -> DAGRunCollectionResponse | ServerResponseError:
@@ -542,10 +547,16 @@ class DagRunOperations(BaseOperations):
start_date: Filter Dag runs by start date (optional)
end_date: Filter Dag runs by end date (optional)
limit: Limit the number of results returned
+ offset: Offset to start returning results from
dag_id: The Dag ID to filter by. If None, retrieves Dag runs for
all Dags (using "~").
logical_date_gte: Filter Dag runs with a logical date greater than
or equal to this value.
logical_date_lte: Filter Dag runs with a logical date less than or
equal to this value.
+ partition_date_gte: Inclusive lower bound of the partition_date
window, as a local
+ calendar day in the Dag's timetable timezone.
+ partition_date_lte: Inclusive upper bound of the partition_date
window, as a local
+ calendar day in the Dag's timetable timezone.
order_by: Order the results by the specified field.
+ partition_key_pattern: Filter Dag runs by partition key pattern.
suppress_error_log: Skip client-side error logging, for callers
handling the error themselves.
"""
# Use "~" for all Dags if dag_id is not specified
@@ -554,12 +565,16 @@ class DagRunOperations(BaseOperations):
params = _build_query_params(
limit=limit,
+ offset=offset,
state=str(state) if state is not None else None,
start_date=start_date,
end_date=end_date,
logical_date_gte=logical_date_gte,
logical_date_lte=logical_date_lte,
+ partition_date_gte=partition_date_gte,
+ partition_date_lte=partition_date_lte,
order_by=order_by,
+ partition_key_pattern=partition_key_pattern,
)
self.response = self.client.get(
diff --git a/airflow-ctl/src/airflowctl/ctl/cli_config.py
b/airflow-ctl/src/airflowctl/ctl/cli_config.py
index 899067d3410..258b565ce1b 100755
--- a/airflow-ctl/src/airflowctl/ctl/cli_config.py
+++ b/airflow-ctl/src/airflowctl/ctl/cli_config.py
@@ -292,6 +292,50 @@ ARG_LOGICAL_DATE = Arg(
type=str,
help="The logical date of the Dag run with a timezone offset (pass this or
run_id, not both)",
)
+ARG_DAG_RUN_ID = Arg(
+ flags=("--run-id",),
+ type=str,
+ help="The Dag run ID to clear",
+)
+ARG_DAG_PARTITION_KEY = Arg(
+ flags=("--partition-key",),
+ type=str,
+ help="The Dag run partition key to clear",
+)
+ARG_DAG_PARTITION_DATE_START = Arg(
+ flags=("--partition-date-start",),
+ type=str,
+ help=(
+ "Inclusive lower bound of the partition_date window, interpreted as a
local calendar "
+ "day in the Dag's timetable timezone. Any time-of-day component is
ignored."
+ ),
+)
+ARG_DAG_PARTITION_DATE_END = Arg(
+ flags=("--partition-date-end",),
+ type=str,
+ help=(
+ "Inclusive upper bound of the partition_date window, interpreted as a
local calendar "
+ "day in the Dag's timetable timezone. Any time-of-day component is
ignored."
+ ),
+)
+ARG_DAG_CLEAR_ONLY_FAILED = Arg(
+ flags=("-f", "--only-failed"),
+ default=False,
+ action="store_true",
+ help="Only clear failed task instances",
+)
+ARG_DAG_CLEAR_ONLY_RUNNING = Arg(
+ flags=("-r", "--only-running"),
+ default=False,
+ action="store_true",
+ help="Only clear running task instances",
+)
+ARG_DAG_CLEAR_YES = Arg(
+ flags=("-y", "--yes"),
+ default=False,
+ action="store_true",
+ help="Do not prompt to confirm clearing task instances",
+)
# Task Commands Args
ARG_RUN_ID = Arg(
@@ -536,6 +580,7 @@ class CommandFactory:
"dict",
"tuple",
"set",
+ "datetime.date",
"datetime.datetime",
}
# Handle Optional types (e.g., "datetime.datetime | None", "str |
None")
@@ -566,6 +611,7 @@ class CommandFactory:
"dict": json_dict_type,
"tuple": tuple,
"set": set,
+ "datetime.date": datetime.date,
"datetime.datetime": datetime.datetime,
"dict[str, typing.Any]": json_dict_type,
}
@@ -1027,6 +1073,21 @@ CONNECTION_COMMANDS = (
)
DAG_COMMANDS = (
+ ActionCommand(
+ name="clear",
+ help="Clear task instances for Dag runs selected by run ID, partition
key, or partition date",
+ func=lazy_load_command("airflowctl.ctl.commands.dag_command.clear"),
+ args=(
+ ARG_DAG_ID,
+ ARG_DAG_RUN_ID,
+ ARG_DAG_PARTITION_KEY,
+ ARG_DAG_PARTITION_DATE_START,
+ ARG_DAG_PARTITION_DATE_END,
+ ARG_DAG_CLEAR_ONLY_FAILED,
+ ARG_DAG_CLEAR_ONLY_RUNNING,
+ ARG_DAG_CLEAR_YES,
+ ),
+ ),
ActionCommand(
name="next-execution",
help="Show the next scheduled execution time for a Dag",
diff --git a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
index d2acee3d24a..15307085f36 100644
--- a/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
+++ b/airflow-ctl/src/airflowctl/ctl/commands/dag_command.py
@@ -32,7 +32,11 @@ from airflowctl.api.client import (
ServerResponseError,
provide_api_client,
)
-from airflowctl.api.datamodels.generated import DAGPatchBody, DAGRunResponse
+from airflowctl.api.datamodels.generated import (
+ ClearTaskInstancesBody,
+ DAGPatchBody,
+ DAGRunResponse,
+)
from airflowctl.ctl.console_formatting import AirflowConsole
@@ -93,17 +97,17 @@ _NEXT_EXECUTION_FIELDS = (
@provide_api_client(kind=ClientKind.CLI)
def next_execution(args, api_client=NEW_API_CLIENT) -> dict | None:
- """Show next scheduled execution time for a DAG."""
+ """Show next scheduled execution time for a Dag."""
try:
response = api_client.dags.get(dag_id=args.dag_id)
except ServerResponseError as e:
- rich.print(f"[red]Error retrieving DAG {args.dag_id}: {e}[/red]")
+ rich.print(f"[red]Error retrieving Dag {args.dag_id}: {e}[/red]")
sys.exit(1)
next_exec_data = {field: getattr(response, field) for field in
_NEXT_EXECUTION_FIELDS}
if all(value is None for value in next_exec_data.values()):
- rich.print(f"[yellow]No upcoming run scheduled for DAG
{args.dag_id}.[/yellow]")
+ rich.print(f"[yellow]No upcoming run scheduled for Dag
{args.dag_id}.[/yellow]")
return None
result = next_exec_data
@@ -172,3 +176,157 @@ def state(args, api_client=NEW_API_CLIENT) -> None:
rich.print(Text(f"{state_value}, {json.dumps(dag_run.conf)}"))
else:
rich.print(Text(state_value))
+
+
+def _parse_partition_date(value: str | None, *, option: str) -> datetime.date
| None:
+ if value is None:
+ return None
+
+ try:
+ if "T" not in value and " " not in value:
+ return datetime.date.fromisoformat(value)
+
+ return datetime.datetime.fromisoformat(value.replace("Z",
"+00:00")).date()
+ except ValueError:
+ raise SystemExit(
+ f"Invalid {option}: {value!r}. Use YYYY-MM-DD or ISO 8601
datetime; only the date is used."
+ ) from None
+
+
+def _parse_and_validate_clear_args(args) -> tuple[datetime.date | None,
datetime.date | None]:
+ """Reject invalid selector combinations and return the parsed
partition_date window."""
+ has_run_id = args.run_id is not None
+ has_partition_key = args.partition_key is not None
+ has_partition_date = args.partition_date_start is not None or
args.partition_date_end is not None
+
+ if sum([has_run_id, has_partition_key, has_partition_date]) != 1:
+ raise SystemExit(
+ "Exactly one selector is required: --run-id, --partition-key, "
+ "or --partition-date-start with --partition-date-end."
+ )
+ if has_partition_date and (args.partition_date_start is None or
args.partition_date_end is None):
+ raise SystemExit("--partition-date-start and --partition-date-end must
be provided together.")
+ if args.only_failed and args.only_running:
+ raise SystemExit("--only-failed and --only-running are mutually
exclusive.")
+
+ partition_date_start = _parse_partition_date(args.partition_date_start,
option="--partition-date-start")
+ partition_date_end = _parse_partition_date(args.partition_date_end,
option="--partition-date-end")
+ if (
+ partition_date_start is not None
+ and partition_date_end is not None
+ and partition_date_start > partition_date_end
+ ):
+ raise SystemExit("--partition-date-start must be before or equal to
--partition-date-end.")
+ return partition_date_start, partition_date_end
+
+
+def _list_dag_runs(api_client, dag_id: str, *, order_by: str = "logical_date",
**filters) -> list:
+ dag_runs = []
+ offset = 0
+ while True:
+ response = api_client.dag_runs.list(
+ dag_id=dag_id,
+ offset=offset,
+ order_by=order_by,
+ **filters,
+ )
+ page_dag_runs = response.dag_runs
+ dag_runs.extend(page_dag_runs)
+
+ offset += len(page_dag_runs)
+ if not page_dag_runs or response.total_entries is None or offset >=
response.total_entries:
+ return dag_runs
+
+
+def _get_dag_runs_to_clear(
+ args,
+ api_client,
+ partition_date_start: datetime.date | None,
+ partition_date_end: datetime.date | None,
+) -> list:
+ if args.run_id is not None:
+ return [api_client.dag_runs.get(dag_id=args.dag_id,
dag_run_id=args.run_id)]
+
+ if args.partition_key is not None:
+ return [
+ dag_run
+ for dag_run in _list_dag_runs(
+ api_client,
+ args.dag_id,
+ order_by="partition_date",
+ partition_key_pattern=args.partition_key,
+ )
+ if dag_run.partition_key == args.partition_key
+ ]
+
+ return _list_dag_runs(
+ api_client,
+ args.dag_id,
+ order_by="partition_date",
+ partition_date_gte=partition_date_start,
+ partition_date_lte=partition_date_end,
+ )
+
+
+def _print_dag_runs_to_clear(dag_id: str, dag_runs: list) -> None:
+ rich.print(f"[yellow]Dag:[/yellow] {dag_id}")
+ rich.print(f"[yellow]Dag runs to clear:[/yellow] {len(dag_runs)}")
+ for dag_run in dag_runs:
+ logical_date = dag_run.logical_date.isoformat() if
dag_run.logical_date is not None else "-"
+ partition_date = getattr(dag_run, "partition_date", None)
+ partition_date_display = partition_date.isoformat() if partition_date
is not None else "-"
+ rich.print(
+ f" - {dag_run.dag_run_id} (logical date: {logical_date},
partition date: {partition_date_display})"
+ )
+
+
+def _confirm_clear(dag_id: str, dag_runs: list) -> bool:
+ _print_dag_runs_to_clear(dag_id, dag_runs)
+ answer = input("Clear task instances for these Dag runs? [y/N] ")
+ return answer.strip().lower() in {"y", "yes"}
+
+
+def _get_dag_run_sort_key(dag_run) -> tuple[str, str, str]:
+ partition_date: datetime.datetime | None = getattr(dag_run,
"partition_date", None)
+ logical_date: datetime.datetime | None = dag_run.logical_date
+ return (
+ partition_date.isoformat() if partition_date is not None else "",
+ logical_date.isoformat() if logical_date is not None else "",
+ dag_run.dag_run_id,
+ )
+
+
+@provide_api_client(kind=ClientKind.CLI)
+def clear(args, api_client=NEW_API_CLIENT) -> dict[str, int | bool]:
+ """Clear task instances for selected Dag runs."""
+ partition_date_start, partition_date_end =
_parse_and_validate_clear_args(args)
+
+ dag_runs = _get_dag_runs_to_clear(args, api_client, partition_date_start,
partition_date_end)
+ if not dag_runs:
+ rich.print(f"[yellow]No matching Dag runs found for
{args.dag_id}.[/yellow]")
+ return {"dag_run_count": 0, "cleared_task_instances": 0}
+
+ dag_runs = sorted(dag_runs, key=_get_dag_run_sort_key)
+
+ if not args.yes and not _confirm_clear(args.dag_id, dag_runs):
+ rich.print("[yellow]Cancelled.[/yellow]")
+ return {"dag_run_count": len(dag_runs), "cleared_task_instances": 0,
"cancelled": True}
+
+ cleared_task_instances = 0
+ for dag_run in dag_runs:
+ response = api_client.tasks.clear(
+ dag_id=args.dag_id,
+ clear_task_instances=ClearTaskInstancesBody(
+ dag_run_id=dag_run.dag_run_id,
+ dry_run=False,
+ only_failed=args.only_failed,
+ only_running=args.only_running,
+ reset_dag_runs=True,
+ ),
+ )
+ cleared_task_instances += response.total_entries or 0
+
+ rich.print(
+ f"[green]Cleared {cleared_task_instances} task instance(s) across
{len(dag_runs)} Dag run(s).[/green]"
+ )
+ return {"dag_run_count": len(dag_runs), "cleared_task_instances":
cleared_task_instances}
diff --git a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml
b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml
index 9c2c591417b..395b8cb3e13 100644
--- a/airflow-ctl/src/airflowctl/ctl/help_texts.yaml
+++ b/airflow-ctl/src/airflowctl/ctl/help_texts.yaml
@@ -52,6 +52,7 @@ connections:
test: "Test connectivity for a given connection"
dags:
+ clear: "Clear task instances for Dag runs selected by run ID, partition key,
or partition date"
get: "Retrieve a Dag by its ID"
get-details: "Retrieve detailed information for a Dag"
get-tags: "List all tags used across Dags"
diff --git a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
index b18beddd677..d20f08e77f4 100644
--- a/airflow-ctl/tests/airflow_ctl/api/test_operations.py
+++ b/airflow-ctl/tests/airflow_ctl/api/test_operations.py
@@ -1301,6 +1301,43 @@ class TestDagRunOperations:
)
assert response == self.dag_run_collection_response
+ def test_list_with_clear_filters(self):
+ logical_date_start = datetime.datetime(2025, 1, 1, 0, 0, 0,
tzinfo=datetime.timezone.utc)
+ logical_date_end = datetime.datetime(2025, 1, 2, 23, 59, 59,
tzinfo=datetime.timezone.utc)
+ partition_day_start = datetime.date(2025, 2, 1)
+ partition_day_end = datetime.date(2025, 2, 4)
+
+ def handle_request(request: httpx.Request) -> httpx.Response:
+ assert request.url.path == f"/api/v2/dags/{self.dag_id}/dagRuns"
+ # Asserted exactly: a name the endpoint does not declare is
silently ignored by
+ # FastAPI, so an unfiltered list would come back and the caller
would act on it.
+ assert dict(request.url.params) == {
+ "limit": "50",
+ "offset": "100",
+ "logical_date_gte": logical_date_start.isoformat(),
+ "logical_date_lte": logical_date_end.isoformat(),
+ "partition_date_gte": partition_day_start.isoformat(),
+ "partition_date_lte": partition_day_end.isoformat(),
+ "order_by": "logical_date",
+ "partition_key_pattern": "2025-01-01",
+ }
+ return httpx.Response(200,
json=json.loads(self.dag_run_collection_response.model_dump_json()))
+
+ client = make_api_client(transport=httpx.MockTransport(handle_request))
+ response = client.dag_runs.list(
+ dag_id=self.dag_id,
+ limit=50,
+ offset=100,
+ logical_date_gte=logical_date_start,
+ logical_date_lte=logical_date_end,
+ partition_date_gte=partition_day_start,
+ partition_date_lte=partition_day_end,
+ order_by="logical_date",
+ partition_key_pattern="2025-01-01",
+ )
+
+ assert response == self.dag_run_collection_response
+
@pytest.mark.parametrize(
(
"dag_id_input",
diff --git a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
index 62c402afb6c..1425868e282 100644
--- a/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
+++ b/airflow-ctl/tests/airflow_ctl/ctl/commands/test_dag_command.py
@@ -17,14 +17,16 @@
from __future__ import annotations
import datetime
+from types import SimpleNamespace
from unittest import mock
+from unittest.mock import Mock, call, create_autospec
import httpx
import pytest
from airflowctl.api.client import ClientKind
-from airflowctl.api.datamodels.generated import DAGResponse
-from airflowctl.api.operations import ServerResponseError
+from airflowctl.api.datamodels.generated import ClearTaskInstancesBody,
DAGResponse
+from airflowctl.api.operations import DagRunOperations, ServerResponseError,
TasksOperations
from airflowctl.ctl import cli_parser
from airflowctl.ctl.commands import dag_command
@@ -133,6 +135,28 @@ class TestDagCommands:
is_stale=False,
)
+ @staticmethod
+ def _dag_run(
+ dag_run_id: str,
+ *,
+ logical_date: datetime.datetime | None = datetime.datetime(2025, 1, 1,
0, 0, 0),
+ partition_key: str | None = None,
+ partition_date: datetime.datetime | None = datetime.datetime(2025, 1,
1, 0, 0, 0),
+ ):
+ return SimpleNamespace(
+ dag_run_id=dag_run_id,
+ logical_date=logical_date,
+ partition_key=partition_key,
+ partition_date=partition_date,
+ )
+
+ @staticmethod
+ def _api_client_mock():
+ api_client = Mock(spec_set=["dag_runs", "tasks"])
+ api_client.dag_runs = create_autospec(DagRunOperations, instance=True,
spec_set=True)
+ api_client.tasks = create_autospec(TasksOperations, instance=True,
spec_set=True)
+ return api_client
+
def test_pause_dag(self, api_client_maker, monkeypatch):
api_client = api_client_maker(
path=f"/api/v2/dags/{self.dag_id}",
@@ -380,3 +404,280 @@ class TestDagCommands:
assert ctx.value is error
api_client.dag_runs.get.assert_not_called()
+
+ def test_clear_by_run_id(self):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.get.return_value =
self._dag_run("scheduled__2025-01-01")
+ api_client.tasks.clear.return_value = SimpleNamespace(total_entries=2)
+
+ result = dag_command.clear(
+ self.parser.parse_args(
+ ["dags", "clear", self.dag_id, "--run-id",
"scheduled__2025-01-01", "--yes"]
+ ),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 1, "cleared_task_instances": 2}
+ api_client.dag_runs.get.assert_called_once_with(
+ dag_id=self.dag_id, dag_run_id="scheduled__2025-01-01"
+ )
+ api_client.tasks.clear.assert_called_once_with(
+ dag_id=self.dag_id,
+ clear_task_instances=ClearTaskInstancesBody(
+ dag_run_id="scheduled__2025-01-01",
+ dry_run=False,
+ only_failed=False,
+ only_running=False,
+ reset_dag_runs=True,
+ ),
+ )
+
+ def test_clear_by_partition_key_filters_exact_match_and_paginates(self):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.list.side_effect = [
+ SimpleNamespace(
+ dag_runs=[
+ self._dag_run(
+ "scheduled__2025-01-01",
+ logical_date=datetime.datetime(2025, 1, 1, 0, 0, 0),
+ partition_key="customer-a",
+ ),
+ self._dag_run(
+ "scheduled__2025-01-02",
+ logical_date=datetime.datetime(2025, 1, 2, 0, 0, 0),
+ partition_key="customer-a-suffix",
+ ),
+ ],
+ total_entries=3,
+ ),
+ SimpleNamespace(
+ dag_runs=[
+ self._dag_run(
+ "scheduled__2025-01-03",
+ logical_date=datetime.datetime(2025, 1, 3, 0, 0, 0),
+ partition_key="customer-a",
+ )
+ ],
+ total_entries=3,
+ ),
+ ]
+ api_client.tasks.clear.side_effect = [
+ SimpleNamespace(total_entries=1),
+ SimpleNamespace(total_entries=2),
+ ]
+
+ result = dag_command.clear(
+ self.parser.parse_args(["dags", "clear", self.dag_id,
"--partition-key", "customer-a", "--yes"]),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 2, "cleared_task_instances": 3}
+ assert api_client.dag_runs.list.call_args_list == [
+ call(
+ dag_id=self.dag_id,
+ offset=0,
+ order_by="partition_date",
+ partition_key_pattern="customer-a",
+ ),
+ call(
+ dag_id=self.dag_id,
+ offset=2,
+ order_by="partition_date",
+ partition_key_pattern="customer-a",
+ ),
+ ]
+ assert api_client.tasks.clear.call_args_list == [
+ call(
+ dag_id=self.dag_id,
+ clear_task_instances=ClearTaskInstancesBody(
+ dag_run_id="scheduled__2025-01-01",
+ dry_run=False,
+ only_failed=False,
+ only_running=False,
+ reset_dag_runs=True,
+ ),
+ ),
+ call(
+ dag_id=self.dag_id,
+ clear_task_instances=ClearTaskInstancesBody(
+ dag_run_id="scheduled__2025-01-03",
+ dry_run=False,
+ only_failed=False,
+ only_running=False,
+ reset_dag_runs=True,
+ ),
+ ),
+ ]
+
+ def test_clear_by_partition_date_uses_partition_date_filters(self):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.list.return_value = SimpleNamespace(
+ dag_runs=[self._dag_run("scheduled__2025-01-01")],
+ total_entries=1,
+ )
+ api_client.tasks.clear.return_value = SimpleNamespace(total_entries=1)
+
+ result = dag_command.clear(
+ self.parser.parse_args(
+ [
+ "dags",
+ "clear",
+ self.dag_id,
+ "--partition-date-start",
+ "2025-01-01",
+ "--partition-date-end",
+ "2025-01-02",
+ "--only-running",
+ "--yes",
+ ]
+ ),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 1, "cleared_task_instances": 1}
+ api_client.dag_runs.list.assert_called_once_with(
+ dag_id=self.dag_id,
+ offset=0,
+ order_by="partition_date",
+ partition_date_gte=datetime.date(2025, 1, 1),
+ partition_date_lte=datetime.date(2025, 1, 2),
+ )
+ api_client.tasks.clear.assert_called_once_with(
+ dag_id=self.dag_id,
+ clear_task_instances=ClearTaskInstancesBody(
+ dag_run_id="scheduled__2025-01-01",
+ dry_run=False,
+ only_failed=False,
+ only_running=True,
+ reset_dag_runs=True,
+ ),
+ )
+
+ def test_clear_by_partition_date_uses_calendar_dates_from_datetimes(self):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.list.return_value = SimpleNamespace(
+ dag_runs=[self._dag_run("scheduled__2025-01-01")],
+ total_entries=1,
+ )
+ api_client.tasks.clear.return_value = SimpleNamespace(total_entries=1)
+
+ result = dag_command.clear(
+ self.parser.parse_args(
+ [
+ "dags",
+ "clear",
+ self.dag_id,
+ "--partition-date-start",
+ "2025-01-01T08:00:00+08:00",
+ "--partition-date-end",
+ "2025-01-02T17:00:00+08:00",
+ "--yes",
+ ]
+ ),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 1, "cleared_task_instances": 1}
+ api_client.dag_runs.list.assert_called_once_with(
+ dag_id=self.dag_id,
+ offset=0,
+ order_by="partition_date",
+ partition_date_gte=datetime.date(2025, 1, 1),
+ partition_date_lte=datetime.date(2025, 1, 2),
+ )
+
+ def
test_clear_by_partition_date_accepts_naive_datetime_as_calendar_date(self):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.list.return_value = SimpleNamespace(
+ dag_runs=[self._dag_run("scheduled__2025-01-01")],
+ total_entries=1,
+ )
+ api_client.tasks.clear.return_value = SimpleNamespace(total_entries=1)
+
+ result = dag_command.clear(
+ self.parser.parse_args(
+ [
+ "dags",
+ "clear",
+ self.dag_id,
+ "--partition-date-start",
+ "2025-01-01T00:00:00",
+ "--partition-date-end",
+ "2025-01-02T00:00:00",
+ "--yes",
+ ]
+ ),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 1, "cleared_task_instances": 1}
+ api_client.dag_runs.list.assert_called_once_with(
+ dag_id=self.dag_id,
+ offset=0,
+ order_by="partition_date",
+ partition_date_gte=datetime.date(2025, 1, 1),
+ partition_date_lte=datetime.date(2025, 1, 2),
+ )
+
+ def test_clear_prompts_before_clearing(self, monkeypatch):
+ api_client = self._api_client_mock()
+ api_client.dag_runs.get.return_value =
self._dag_run("scheduled__2025-01-01")
+ monkeypatch.setattr("builtins.input", lambda _: "n")
+
+ result = dag_command.clear(
+ self.parser.parse_args(["dags", "clear", self.dag_id, "--run-id",
"scheduled__2025-01-01"]),
+ api_client=api_client,
+ )
+
+ assert result == {"dag_run_count": 1, "cleared_task_instances": 0,
"cancelled": True}
+ api_client.tasks.clear.assert_not_called()
+
+ @pytest.mark.parametrize(
+ "command",
+ [
+ ["dags", "clear", dag_id],
+ ["dags", "clear", dag_id, "--run-id", "run", "--partition-key",
"key"],
+ [
+ "dags",
+ "clear",
+ dag_id,
+ "--partition-date-start",
+ "2025-01-01",
+ ],
+ [
+ "dags",
+ "clear",
+ dag_id,
+ "--run-id",
+ "run",
+ "--only-failed",
+ "--only-running",
+ ],
+ [
+ "dags",
+ "clear",
+ dag_id,
+ "--partition-date-start",
+ "2025-01-02",
+ "--partition-date-end",
+ "2025-01-01",
+ ],
+ [
+ "dags",
+ "clear",
+ dag_id,
+ "--partition-date-start",
+ "not-a-date",
+ "--partition-date-end",
+ "2025-01-01",
+ ],
+ ],
+ )
+ def test_clear_validates_selectors(self, command):
+ api_client = self._api_client_mock()
+
+ with pytest.raises(SystemExit):
+ dag_command.clear(self.parser.parse_args(command),
api_client=api_client)
+
+ api_client.dag_runs.list.assert_not_called()
+ api_client.tasks.clear.assert_not_called()
