Merge branch 'main' into issue-50000

Author: shubhamraj-git

.github/CODEOWNERS

@@ -37,7 +37,7 @@
 
 # Translations
 airflow-core/src/airflow/ui/src/i18n/locales/de/ @jscheffl
-airflow-core/src/airflow/ui/src/i18n/locales/zh_TW/ @Lee-W
+airflow-core/src/airflow/ui/src/i18n/locales/zh-TW/ @Lee-W
 airflow-core/src/airflow/ui/src/i18n/locales/nl/ @BasPH # not codeowner but engaged: @DjVinnii
 
 # Security/Permissions

ISSUE_TRIAGE_PROCESS.rst

@@ -196,7 +196,7 @@ associated with them such as ``provider:amazon-aws``, ``provider:microsoft-azure
 These make it easier for developers working on a single provider to
 track issues for that provider.
 
-Note: each provider has it's own unique label. It is possible for issue to be tagged with more than 1 provider label.
+Note: each provider has its own unique label. It is possible for issue to be tagged with more than 1 provider label.
 
 Most issues need a combination of "kind" and "area" labels to be actionable.
 For example:

airflow-core/docs/tutorial/fundamentals.rst

@@ -43,7 +43,7 @@ each line in detail.
 Understanding the DAG Definition File
 -------------------------------------
 Think of the Airflow Python script as a configuration file that lays out the structure of your DAG in code. The actual
-tasks you define here run in a different environment, which means this script isn't meant for data processing. It's main
+tasks you define here run in a different environment, which means this script isn't meant for data processing. Its main
 job is to define the DAG object, and it needs to evaluate quickly since the DAG File Processor checks it regularly for
 any changes.
 

airflow-core/src/airflow/api_fastapi/core_api/routes/public/dags.py

@@ -22,12 +22,13 @@
 from fastapi import Depends, HTTPException, Query, Response, status
 from fastapi.exceptions import RequestValidationError
 from pydantic import ValidationError
-from sqlalchemy import select, update
+from sqlalchemy import func, null, select, update
 
 from airflow.api.common import delete_dag as delete_dag_module
 from airflow.api_fastapi.common.dagbag import DagBagDep
 from airflow.api_fastapi.common.db.common import (
     SessionDep,
+    apply_filters_to_select,
     paginated_select,
 )
 from airflow.api_fastapi.common.db.dags import generate_dag_with_latest_run_query
@@ -115,30 +116,54 @@ def get_dags(
     session: SessionDep,
 ) -> DAGCollectionResponse:
     """Get all DAGs."""
-    dag_runs_select = None
+    query = select(DagModel)
 
-    if dag_run_state.value or dag_run_start_date_range.is_active() or dag_run_end_date_range.is_active():
-        dag_runs_select, _ = paginated_select(
-            statement=select(DagRun),
+    max_run_id_query = (  # ordering by id will not always be "latest run", but it's a simplifying assumption
+        select(DagRun.dag_id, func.max(DagRun.id).label("max_dag_run_id"))
+        .where(DagRun.start_date.is_not(null()))
+        .group_by(DagRun.dag_id)
+        .subquery(name="mrq")
+    )
+
+    has_max_run_filter = (
+        dag_run_state.value
+        or last_dag_run_state.value
+        or dag_run_start_date_range.is_active()
+        or dag_run_end_date_range.is_active()
+    )
+
+    if has_max_run_filter or order_by.value in (
+        "last_run_state",
+        "last_run_start_date",
+        "-last_run_state",
+        "-last_run_start_date",
+    ):
+        query = query.join(
+            max_run_id_query,
+            DagModel.dag_id == max_run_id_query.c.dag_id,
+            isouter=True,
+        ).join(DagRun, DagRun.id == max_run_id_query.c.max_dag_run_id, isouter=True)
+
+    if has_max_run_filter:
+        query = apply_filters_to_select(
+            statement=query,
             filters=[
                 dag_run_start_date_range,
                 dag_run_end_date_range,
                 dag_run_state,
+                last_dag_run_state,
             ],
-            session=session,
         )
-        dag_runs_select = dag_runs_select.cte()
 
     dags_select, total_entries = paginated_select(
-        statement=generate_dag_with_latest_run_query(dag_runs_select),
+        statement=query,
         filters=[
             exclude_stale,
             paused,
             dag_id_pattern,
             dag_display_name_pattern,
             tags,
             owners,
-            last_dag_run_state,
             readable_dags_filter,
         ],
         order_by=order_by,

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dags.py

@@ -172,7 +172,7 @@ class TestGetDags(TestDagEndpoint):
             ({"last_dag_run_state": "success", "exclude_stale": False}, 1, [DAG3_ID]),
             ({"last_dag_run_state": "failed", "exclude_stale": False}, 1, [DAG1_ID]),
             ({"dag_run_state": "failed"}, 1, [DAG1_ID]),
-            ({"dag_run_state": "failed", "exclude_stale": False}, 2, [DAG1_ID, DAG3_ID]),
+            ({"dag_run_state": "failed", "exclude_stale": False}, 1, [DAG1_ID]),
             (
                 {"dag_run_start_date_gte": DAG3_START_DATE_2.isoformat(), "exclude_stale": False},
                 1,
@@ -220,10 +220,10 @@ class TestGetDags(TestDagEndpoint):
                     "dag_run_state": "failed",
                     "exclude_stale": False,
                 },
-                1,
-                [DAG3_ID],
+                0,
+                [],
             ),
-            # # Sort
+            # Sort
             ({"order_by": "-dag_id"}, 2, [DAG2_ID, DAG1_ID]),
             ({"order_by": "-dag_display_name"}, 2, [DAG2_ID, DAG1_ID]),
             ({"order_by": "dag_display_name"}, 2, [DAG1_ID, DAG2_ID]),

airflow-ctl/docs/images/command_hashes.txt

@@ -9,6 +9,6 @@ dagrun:7b3e06a3664cc7ceb18457b4c0895532
 jobs:806174e6c9511db669705279ed6a00b9
 pools:2c17a4131b6481bd8fe9120982606db2
 providers:d053e6f17ff271e1e08942378344d27b
-variables:d9001295d77adefbd68e389f6622b89a
+variables:cd3970589b2cb1e3ebd9a0b7f2ffdf4d
 version:11da98f530c37754403a87151cbe2274
 auth login:348c25d49128b6007ac97dae2ef7563f

airflow-ctl/docs/images/output_variables.svg

@@ -198,6 +198,23 @@ def __call__(self, parser, namespace, values, option_string=None):
     action=Password,
     nargs="?",
 )
+ARG_VARIABLE_IMPORT = Arg(
+    flags=("file",),
+    metavar="file",
+    help="Import variables from JSON file",
+)
+ARG_VARIABLE_ACTION_ON_EXISTING_KEY = Arg(
+    flags=("-a", "--action-on-existing-key"),
+    type=str,
+    default="overwrite",
+    help="Action to take if we encounter a variable key that already exists.",
+    choices=("overwrite", "fail", "skip"),
+)
+ARG_VARIABLE_EXPORT = Arg(
+    flags=("file",),
+    metavar="file",
+    help="Export all variables to JSON file",
+)
 
 ARG_OUTPUT = Arg(
     flags=("-o", "--output"),
@@ -626,6 +643,21 @@ def merge_commands(
     ),
 )
 
+VARIABLE_COMMANDS = (
+    ActionCommand(
+        name="import",
+        help="Import variables",
+        func=lazy_load_command("airflowctl.ctl.commands.variable_command.import_"),
+        args=(ARG_VARIABLE_IMPORT, ARG_VARIABLE_ACTION_ON_EXISTING_KEY),
+    ),
+    ActionCommand(
+        name="export",
+        help="Export all variables",
+        func=lazy_load_command("airflowctl.ctl.commands.variable_command.export"),
+        args=(ARG_VARIABLE_EXPORT,),
+    ),
+)
+
 core_commands: list[CLICommand] = [
     GroupCommand(
         name="auth",
@@ -638,6 +670,11 @@ def merge_commands(
         help="Manage Airflow pools",
         subcommands=POOL_COMMANDS,
     ),
+    GroupCommand(
+        name="variables",
+        help="Manage Airflow variables",
+        subcommands=VARIABLE_COMMANDS,
+    ),
 ]
 # Add generated group commands
 core_commands = merge_commands(