diff --git a/airflow-core/src/airflow/models/dagrun.py b/airflow-core/src/airflow/models/dagrun.py
index 7d41528df5ca7..5bdefb99f775e 100644
--- a/airflow-core/src/airflow/models/dagrun.py
+++ b/airflow-core/src/airflow/models/dagrun.py
@@ -1195,27 +1195,31 @@ def get_previous_scheduled_dagrun(
def _tis_for_dagrun_state(self, *, dag, tis):
"""
- Return the collection of tasks that should be considered for evaluation of terminal dag run state.
-
- Teardown tasks by default are not considered for the purpose of dag run state. But
- users may enable such consideration with on_failure_fail_dagrun.
+ Return the collection of tasks that should be considered for evaluation
+ of terminal dag run state.
"""
def is_effective_leaf(task):
for down_task_id in task.downstream_task_ids:
down_task = dag.get_task(down_task_id)
- if not down_task.is_teardown or down_task.on_failure_fail_dagrun:
- # we found a down task that is not ignorable; not a leaf
+
+ if (
+ not down_task.is_teardown
+ or (
+ down_task.on_failure_fail_dagrun
+ and down_task.on_success_succeed_dagrun
+ )
+ ):
return False
- # we found no ignorable downstreams
- # evaluate whether task is itself ignorable
return not task.is_teardown or task.on_failure_fail_dagrun
-
leaf_task_ids = {x.task_id for x in dag.tasks if is_effective_leaf(x)}
if not leaf_task_ids:
- # can happen if dag is exclusively teardown tasks
leaf_task_ids = {x.task_id for x in dag.tasks if not x.downstream_list}
- leaf_tis = {ti for ti in tis if ti.task_id in leaf_task_ids if ti.state != TaskInstanceState.REMOVED}
+ leaf_tis = {
+ ti
+ for ti in tis
+ if ti.task_id in leaf_task_ids and ti.state != TaskInstanceState.REMOVED
+ }
return leaf_tis
def _emit_dagrun_span(self, state: DagRunState):
diff --git a/airflow-core/src/airflow/ui/public/i18n/locales/en/admin.json b/airflow-core/src/airflow/ui/public/i18n/locales/en/admin.json
index e8735ea2713c0..2656a817a9d15 100644
--- a/airflow-core/src/airflow/ui/public/i18n/locales/en/admin.json
+++ b/airflow-core/src/airflow/ui/public/i18n/locales/en/admin.json
@@ -15,6 +15,7 @@
},
"connections": {
"add": "Add $t(connections.connection_one)",
+ "duplicate": "Duplicate",
"columns": {
"connectionId": "Connection ID",
"connectionType": "Connection Type",
diff --git a/airflow-core/src/airflow/ui/src/pages/Connections/Connections.tsx b/airflow-core/src/airflow/ui/src/pages/Connections/Connections.tsx
index 6279938f5747d..da4bb91ade634 100644
--- a/airflow-core/src/airflow/ui/src/pages/Connections/Connections.tsx
+++ b/airflow-core/src/airflow/ui/src/pages/Connections/Connections.tsx
@@ -52,6 +52,7 @@ import { formatNumber, useDocumentTitle } from "src/utils";
import AddConnectionButton from "./AddConnectionButton";
import DeleteConnectionButton from "./DeleteConnectionButton";
import DeleteConnectionsButton from "./DeleteConnectionsButton";
+import DuplicateConnectionButton from "./DuplicateConnectionButton";
import EditConnectionButton from "./EditConnectionButton";
import { NothingFoundInfo } from "./NothingFoundInfo";
import TestConnectionButton from "./TestConnectionButton";
@@ -125,6 +126,7 @@ const getColumns = ({
+
),
diff --git a/airflow-core/src/airflow/ui/src/pages/Connections/DuplicateConnectionButton.tsx b/airflow-core/src/airflow/ui/src/pages/Connections/DuplicateConnectionButton.tsx
new file mode 100644
index 0000000000000..0af1480883a9d
--- /dev/null
+++ b/airflow-core/src/airflow/ui/src/pages/Connections/DuplicateConnectionButton.tsx
@@ -0,0 +1,87 @@
+
+/*!
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { useDisclosure } from "@chakra-ui/react";
+import { useTranslation } from "react-i18next";
+import { FiCopy } from "react-icons/fi";
+
+import type { ConnectionResponse } from "openapi/requests/types.gen";
+
+import { IconButton, Modal } from "src/system-components";
+
+import { useAddConnection } from "src/queries/useAddConnection";
+
+import ConnectionForm from "./ConnectionForm";
+import type { ConnectionBody } from "./Connections";
+
+type Props = {
+ readonly connection: ConnectionResponse;
+ readonly disabled: boolean;
+};
+
+const DuplicateConnectionButton = ({ connection, disabled }: Props) => {
+ const { t: translate } = useTranslation("admin");
+ const { onClose, onOpen, open } = useDisclosure();
+
+ const initialConnection: ConnectionBody = {
+ conn_type: connection.conn_type,
+ connection_id: "",
+ description: connection.description ?? "",
+ extra: connection.extra === "" || connection.extra === null ? "{}" : connection.extra,
+ host: connection.host ?? "",
+ login: connection.login ?? "",
+ password: "",
+ port: connection.port?.toString() ?? "",
+ schema: connection.schema ?? "",
+ team_name: connection.team_name ?? "",
+ };
+
+ const { addConnection, error, isPending } = useAddConnection({
+ onSuccessConfirm: onClose,
+ });
+
+ return (
+ <>
+
+
+
+
+
+
+
+
+ );
+};
+
+export default DuplicateConnectionButton;
diff --git a/airflow-core/tests/unit/models/test_dagrun.py b/airflow-core/tests/unit/models/test_dagrun.py
index 4e00a1392ea3d..699e386fa77b6 100644
--- a/airflow-core/tests/unit/models/test_dagrun.py
+++ b/airflow-core/tests/unit/models/test_dagrun.py
@@ -4784,7 +4784,48 @@ def mytask():
dr = session.scalar(select(DagRun))
assert dr.state == DagRunState.FAILED
+def test_work_task_failure_when_teardown_succeeds_and_on_success_is_false(
+ dag_maker, session
+):
+ with dag_maker():
+
+ @setup
+ def setuptask():
+ pass
+ @teardown(
+ on_failure_fail_dagrun=True,
+ on_success_succeed_dagrun=False,
+ )
+ def teardown_task():
+ pass
+
+ @task
+ def mytask():
+ pass
+
+ with setuptask() >> teardown_task():
+ mytask()
+
+ dr = dag_maker.create_dagrun()
+ s1 = dr.get_task_instance(task_id="setuptask")
+ td1 = dr.get_task_instance(task_id="teardown_task")
+ t1 = dr.get_task_instance(task_id="mytask")
+
+ s1.state = TaskInstanceState.SUCCESS
+ td1.state = TaskInstanceState.SUCCESS
+ t1.state = TaskInstanceState.FAILED
+
+ session.merge(s1)
+ session.merge(td1)
+ session.merge(t1)
+ session.flush()
+
+ dr.update_state()
+ session.flush()
+
+ dr = session.scalar(select(DagRun))
+ assert dr.state == DagRunState.FAILED
def test_failure_of_leaf_task_not_connected_to_teardown_task(dag_maker, session):
with dag_maker():
diff --git a/task-sdk/src/airflow/sdk/bases/decorator.py b/task-sdk/src/airflow/sdk/bases/decorator.py
index 896bedb3abc37..ac3e788460c91 100644
--- a/task-sdk/src/airflow/sdk/bases/decorator.py
+++ b/task-sdk/src/airflow/sdk/bases/decorator.py
@@ -521,7 +521,7 @@ class _TaskDecorator(ExpandableFactory, Generic[FParams, FReturn, OperatorSubcla
is_teardown: bool = False
returns_dag_result: bool = False
on_failure_fail_dagrun: bool = False
-
+ on_success_succeed_dagrun: bool = True
# This is set in __attrs_post_init__ by update_wrapper. Provided here for type hints.
__wrapped__: Callable[FParams, FReturn] = attr.ib(init=False)
@@ -566,6 +566,7 @@ def __call__(self, *args: FParams.args, **kwargs: FParams.kwargs) -> XComArg:
raise ValueError("Trigger rule not configurable for teardown tasks.")
self.kwargs.update(trigger_rule=TriggerRule.ALL_DONE_SETUP_SUCCESS)
on_failure_fail_dagrun = self.kwargs.pop("on_failure_fail_dagrun", self.on_failure_fail_dagrun)
+ on_success_succeed_dagrun = self.kwargs.pop("on_success_succeed_dagrun", self.on_success_succeed_dagrun)
op = self.operator_class(
python_callable=self.function,
op_args=args,
@@ -576,6 +577,7 @@ def __call__(self, *args: FParams.args, **kwargs: FParams.kwargs) -> XComArg:
op.is_setup = self.is_setup
op.is_teardown = self.is_teardown
op.on_failure_fail_dagrun = on_failure_fail_dagrun
+ op.on_success_succeed_dagrun = on_success_succeed_dagrun
if self.returns_dag_result:
check_dag_result_outside_loop(op)
op.returns_dag_result = self.returns_dag_result
@@ -660,6 +662,7 @@ def _expand(self, expand_input: ExpandInput, *, strict: bool) -> XComArg:
"is_setup": self.is_setup,
"is_teardown": self.is_teardown,
"on_failure_fail_dagrun": self.on_failure_fail_dagrun,
+ "on_success_succeed_dagrun": self.on_success_succeed_dagrun,
}
base_signature = inspect.signature(BaseOperator)
ignore = {
@@ -762,6 +765,7 @@ def override(self, **kwargs: Any) -> _TaskDecorator[FParams, FReturn, OperatorSu
setattr(result, "is_setup", self.is_setup)
setattr(result, "is_teardown", self.is_teardown)
setattr(result, "on_failure_fail_dagrun", self.on_failure_fail_dagrun)
+ setattr(result, "on_success_succeed_dagrun", self.on_success_succeed_dagrun)
return result
diff --git a/task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py b/task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py
index 0cdd8b52a5f37..f6900ec5d03f3 100644
--- a/task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py
+++ b/task-sdk/src/airflow/sdk/definitions/decorators/setup_teardown.py
@@ -58,13 +58,14 @@ def initialize_context(...):
return func
-def teardown_task(_func=None, *, on_failure_fail_dagrun: bool = False) -> Callable:
+def teardown_task(_func=None, *, on_failure_fail_dagrun: bool = False, on_success_succeed_dagrun: bool = True,) -> Callable:
"""
Decorate a function to mark it as a teardown task.
A teardown task runs after all main tasks in its Dag or TaskGroup context.
If ``on_failure_fail_dagrun=True``, a failure in teardown will mark the Dag run as failed.
-
+ If ``on_success_succeed_dagrun=True``, a successful teardown can mark the Dag run as successful
+ even if another task failed.
Example::
@teardown(on_failure_fail_dagrun=True)
@@ -82,6 +83,7 @@ def teardown(func: Callable) -> Callable:
func.is_teardown = True
func.on_failure_fail_dagrun = on_failure_fail_dagrun
+ func.on_success_succeed_dagrun = on_success_succeed_dagrun
return func
if _func is None: