From c6217264f8ca6493c889eac526cc16877072dc25 Mon Sep 17 00:00:00 2001
From: DaveT1991 <dawoodaljanaby@gmail.com>
Date: Thu, 9 Apr 2026 23:32:01 -0400
Subject: [PATCH 1/2] Skip notifier templating with empty template_fields

---
 task-sdk/src/airflow/sdk/bases/notifier.py    |  2 ++
 .../tests/task_sdk/bases/test_notifier.py     | 21 +++++++++++++++++++
 2 files changed, 23 insertions(+)

diff --git a/task-sdk/src/airflow/sdk/bases/notifier.py b/task-sdk/src/airflow/sdk/bases/notifier.py
index 7cfed6ae97efb..198a5c8fa016e 100644
--- a/task-sdk/src/airflow/sdk/bases/notifier.py
+++ b/task-sdk/src/airflow/sdk/bases/notifier.py
@@ -87,6 +87,8 @@ def render_template_fields(
         :param context: Context dict with values to apply on content.
         :param jinja_env: Jinja environment to use for rendering.
         """
+        if not self.template_fields:
+            return
         dag = context.get("dag")
         if not jinja_env:
             jinja_env = self.get_template_env(dag=dag)
diff --git a/task-sdk/tests/task_sdk/bases/test_notifier.py b/task-sdk/tests/task_sdk/bases/test_notifier.py
index b8cedaa518831..bcbca90b10663 100644
--- a/task-sdk/tests/task_sdk/bases/test_notifier.py
+++ b/task-sdk/tests/task_sdk/bases/test_notifier.py
@@ -24,6 +24,7 @@
 import pytest
 
 from airflow.providers.standard.operators.empty import EmptyOperator
+from airflow.serialization.serialized_objects import SerializedDAG
 from airflow.sdk.bases.notifier import BaseNotifier
 from airflow.sdk.definitions.dag import DAG
 
@@ -45,6 +46,13 @@ def notify(self, context: Context) -> None:
         pass
 
 
+class NoTemplateNotifier(BaseNotifier):
+    """Notifier used to verify callbacks without templated fields."""
+
+    def notify(self, context: Context) -> None:
+        pass
+
+
 class TestBaseNotifier:
     def test_render_message_with_message(self):
         with DAG("test_render_message_with_message") as dag:
@@ -97,3 +105,16 @@ def test_notifier_call_with_prepared_context(self, caplog):
             }
         )
         assert notifier.message == "task: some_task"
+
+    def test_notifier_call_with_serialized_dag_and_no_template_fields(self):
+        with DAG("test_notifier_call_with_serialized_dag_and_no_template_fields") as dag:
+            EmptyOperator(task_id="test_id")
+
+        notifier = NoTemplateNotifier()
+        notifier.notify = MagicMock()
+        serialized_dag = SerializedDAG.deserialize_dag(SerializedDAG.serialize_dag(dag))
+        context: Context = {"dag": serialized_dag}
+
+        notifier(context)
+
+        notifier.notify.assert_called_once_with({"dag": serialized_dag})

From a67e9a73c76717ea252d06b33a607f49f442989a Mon Sep 17 00:00:00 2001
From: DaveT1991 <dawoodaljanaby@gmail.com>
Date: Thu, 9 Apr 2026 23:37:12 -0400
Subject: [PATCH 2/2] Handle serialized DAGs in notifier templating

---
 task-sdk/src/airflow/sdk/bases/notifier.py     |  5 ++++-
 task-sdk/tests/task_sdk/bases/test_notifier.py | 12 ++++++++++++
 2 files changed, 16 insertions(+), 1 deletion(-)

diff --git a/task-sdk/src/airflow/sdk/bases/notifier.py b/task-sdk/src/airflow/sdk/bases/notifier.py
index 198a5c8fa016e..809dd1442baaf 100644
--- a/task-sdk/src/airflow/sdk/bases/notifier.py
+++ b/task-sdk/src/airflow/sdk/bases/notifier.py
@@ -91,7 +91,10 @@ def render_template_fields(
             return
         dag = context.get("dag")
         if not jinja_env:
-            jinja_env = self.get_template_env(dag=dag)
+            if dag is not None and hasattr(dag, "get_template_env"):
+                jinja_env = self.get_template_env(dag=dag)
+            else:
+                jinja_env = self.get_template_env()
         self._do_render_template_fields(self, self.template_fields, context, jinja_env, set())
 
     async def async_notify(self, context: Context) -> None:
diff --git a/task-sdk/tests/task_sdk/bases/test_notifier.py b/task-sdk/tests/task_sdk/bases/test_notifier.py
index bcbca90b10663..99927d33359fb 100644
--- a/task-sdk/tests/task_sdk/bases/test_notifier.py
+++ b/task-sdk/tests/task_sdk/bases/test_notifier.py
@@ -118,3 +118,15 @@ def test_notifier_call_with_serialized_dag_and_no_template_fields(self):
         notifier(context)
 
         notifier.notify.assert_called_once_with({"dag": serialized_dag})
+
+    def test_notifier_render_template_fields_with_serialized_dag(self):
+        with DAG("test_notifier_render_template_fields_with_serialized_dag") as dag:
+            EmptyOperator(task_id="test_id")
+
+        notifier = MockNotifier(message="Hello {{ dag.dag_id }}")
+        serialized_dag = SerializedDAG.deserialize_dag(SerializedDAG.serialize_dag(dag))
+        context: Context = {"dag": serialized_dag}
+
+        notifier.render_template_fields(context)
+
+        assert notifier.message == "Hello test_notifier_render_template_fields_with_serialized_dag"