From 79f5856df9e56e91ad09c30c800fd9ed53700422 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Mon, 7 Oct 2024 13:47:01 -0700
Subject: [PATCH 1/8] Follow-up SLA purge

---
 .../endpoints/task_instance_endpoint.py       |  78 +-----------
 .../api_connexion/schemas/sla_miss_schema.py  |  38 ------
 .../schemas/task_instance_schema.py           |   2 -
 airflow/example_dags/tutorial.py              |   1 -
 airflow/models/__init__.py                    |   3 -
 airflow/models/slamiss.py                     |  46 -------
 airflow/notifications/basenotifier.py         |   9 +-
 airflow/utils/context.py                      |   4 +
 airflow/www/auth.py                           |   4 +-
 airflow/www/extensions/init_views.py          |   3 -
 airflow/www/views.py                          | 115 +-----------------
 .../test_mapped_task_instance_endpoint.py     |   4 +-
 .../endpoints/test_task_instance_endpoint.py  |  29 +----
 .../schemas/test_task_instance_schema.py      |  62 +---------
 tests/jobs/test_scheduler_job.py              |   3 -
 .../test_dag_import_error_listener.py         |   2 -
 tests/notifications/test_basenotifier.py      |   1 -
 .../test_task_instance_endpoint.py            |   3 +-
 .../providers/smtp/notifications/test_smtp.py |  33 -----
 tests/test_utils/db.py                        |   7 --
 20 files changed, 22 insertions(+), 425 deletions(-)
 delete mode 100644 airflow/api_connexion/schemas/sla_miss_schema.py
 delete mode 100644 airflow/models/slamiss.py

diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index 23e5f5a6d16f4..7980012488e18 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -48,7 +48,6 @@
 from airflow.api_connexion.security import get_readable_dags
 from airflow.auth.managers.models.resource_details import DagAccessEntity, DagDetails
 from airflow.exceptions import TaskNotFound
-from airflow.models import SlaMiss
 from airflow.models.dagrun import DagRun as DR
 from airflow.models.taskinstance import TaskInstance as TI, clear_task_instances
 from airflow.models.taskinstancehistory import TaskInstanceHistory as TIH
@@ -84,15 +83,6 @@ def get_task_instance(
         select(TI)
         .where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id == task_id)
         .join(TI.dag_run)
-        .outerjoin(
-            SlaMiss,
-            and_(
-                SlaMiss.dag_id == TI.dag_id,
-                SlaMiss.execution_date == DR.execution_date,
-                SlaMiss.task_id == TI.task_id,
-            ),
-        )
-        .add_columns(SlaMiss)
         .options(joinedload(TI.rendered_task_instance_fields))
     )
 
@@ -127,15 +117,6 @@ def get_mapped_task_instance(
         select(TI)
         .where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id == task_id, TI.map_index == map_index)
         .join(TI.dag_run)
-        .outerjoin(
-            SlaMiss,
-            and_(
-                SlaMiss.dag_id == TI.dag_id,
-                SlaMiss.execution_date == DR.execution_date,
-                SlaMiss.task_id == TI.task_id,
-            ),
-        )
-        .add_columns(SlaMiss)
         .options(joinedload(TI.rendered_task_instance_fields))
     )
     task_instance = session.execute(query).one_or_none()
@@ -232,27 +213,12 @@ def get_mapped_task_instances(
     # Count elements before joining extra columns
     total_entries = get_query_count(base_query, session=session)
 
-    # Add SLA miss
-    entry_query = (
-        base_query.outerjoin(
-            SlaMiss,
-            and_(
-                SlaMiss.dag_id == TI.dag_id,
-                SlaMiss.task_id == TI.task_id,
-                SlaMiss.execution_date == DR.execution_date,
-            ),
-        )
-        .add_columns(SlaMiss)
-        .options(joinedload(TI.rendered_task_instance_fields))
-    )
-
     try:
         order_by_params = _get_order_by_params(order_by)
-        entry_query = entry_query.order_by(*order_by_params)
+        entry_query = base_query.order_by(*order_by_params)
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
     task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
@@ -384,27 +350,12 @@ def get_task_instances(
     # Count elements before joining extra columns
     total_entries = get_query_count(base_query, session=session)
 
-    # Add join
-    entry_query = (
-        base_query.outerjoin(
-            SlaMiss,
-            and_(
-                SlaMiss.dag_id == TI.dag_id,
-                SlaMiss.task_id == TI.task_id,
-                SlaMiss.execution_date == DR.execution_date,
-            ),
-        )
-        .add_columns(SlaMiss)
-        .options(joinedload(TI.rendered_task_instance_fields))
-    )
-
     try:
         order_by_params = _get_order_by_params(order_by)
-        entry_query = entry_query.order_by(*order_by_params)
+        entry_query = base_query.order_by(*order_by_params)
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
     task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
@@ -463,16 +414,7 @@ def get_task_instances_batch(session: Session = NEW_SESSION) -> APIResponse:
 
     # Count elements before joining extra columns
     total_entries = get_query_count(base_query, session=session)
-    # Add join
-    base_query = base_query.join(
-        SlaMiss,
-        and_(
-            SlaMiss.dag_id == TI.dag_id,
-            SlaMiss.task_id == TI.task_id,
-            SlaMiss.execution_date == DR.execution_date,
-        ),
-        isouter=True,
-    ).add_columns(SlaMiss)
+
     ti_query = base_query.options(
         joinedload(TI.rendered_task_instance_fields), joinedload(TI.task_instance_note)
     )
@@ -483,7 +425,6 @@ def get_task_instances_batch(session: Session = NEW_SESSION) -> APIResponse:
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    # using execute because we want the SlaMiss entity. Scalars don't return None for missing entities
     task_instances = session.execute(ti_query).all()
 
     return task_instance_collection_schema.dump(
@@ -690,15 +631,6 @@ def set_task_instance_note(
         select(TI)
         .where(TI.dag_id == dag_id, TI.run_id == dag_run_id, TI.task_id == task_id)
         .join(TI.dag_run)
-        .outerjoin(
-            SlaMiss,
-            and_(
-                SlaMiss.dag_id == TI.dag_id,
-                SlaMiss.execution_date == DR.execution_date,
-                SlaMiss.task_id == TI.task_id,
-            ),
-        )
-        .add_columns(SlaMiss)
         .options(joinedload(TI.rendered_task_instance_fields))
     )
     if map_index == -1:
@@ -716,7 +648,7 @@ def set_task_instance_note(
         error_message = f"Task Instance not found for dag_id={dag_id}, run_id={dag_run_id}, task_id={task_id}"
         raise NotFound(error_message)
 
-    ti, sla_miss = result
+    ti = result[0]
 
     current_user_id = get_auth_manager().get_user_id()
     if ti.task_instance_note is None:
@@ -725,7 +657,7 @@ def set_task_instance_note(
         ti.task_instance_note.content = new_note
         ti.task_instance_note.user_id = current_user_id
     session.commit()
-    return task_instance_schema.dump((ti, sla_miss))
+    return task_instance_schema.dump((ti,))
 
 
 @security.requires_access_dag("GET", DagAccessEntity.TASK_INSTANCE)
diff --git a/airflow/api_connexion/schemas/sla_miss_schema.py b/airflow/api_connexion/schemas/sla_miss_schema.py
deleted file mode 100644
index 97a462e186d59..0000000000000
--- a/airflow/api_connexion/schemas/sla_miss_schema.py
+++ /dev/null
@@ -1,38 +0,0 @@
-# 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.
-from __future__ import annotations
-
-from marshmallow_sqlalchemy import SQLAlchemySchema, auto_field
-
-from airflow.models import SlaMiss
-
-
-class SlaMissSchema(SQLAlchemySchema):
-    """Sla Miss Schema."""
-
-    class Meta:
-        """Meta."""
-
-        model = SlaMiss
-
-    task_id = auto_field(dump_only=True)
-    dag_id = auto_field(dump_only=True)
-    execution_date = auto_field(dump_only=True)
-    email_sent = auto_field(dump_only=True)
-    timestamp = auto_field(dump_only=True)
-    description = auto_field(dump_only=True)
-    notification_sent = auto_field(dump_only=True)
diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 0c1daf6ce2c7d..08b7cfdbac9cb 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -26,7 +26,6 @@
 from airflow.api_connexion.schemas.common_schema import JsonObjectField
 from airflow.api_connexion.schemas.enum_schemas import TaskInstanceStateField
 from airflow.api_connexion.schemas.job_schema import JobSchema
-from airflow.api_connexion.schemas.sla_miss_schema import SlaMissSchema
 from airflow.api_connexion.schemas.trigger_schema import TriggerSchema
 from airflow.models import TaskInstance
 from airflow.models.taskinstancehistory import TaskInstanceHistory
@@ -69,7 +68,6 @@ class Meta:
     executor = auto_field()
     executor_config = auto_field()
     note = auto_field()
-    sla_miss = fields.Nested(SlaMissSchema, dump_default=None)
     rendered_map_index = auto_field()
     rendered_fields = JsonObjectField(dump_default={})
     trigger = fields.Nested(TriggerSchema)
diff --git a/airflow/example_dags/tutorial.py b/airflow/example_dags/tutorial.py
index 0e31775c7a9a7..43edd27788121 100644
--- a/airflow/example_dags/tutorial.py
+++ b/airflow/example_dags/tutorial.py
@@ -55,7 +55,6 @@
         # 'priority_weight': 10,
         # 'end_date': datetime(2016, 1, 1),
         # 'wait_for_downstream': False,
-        # 'sla': timedelta(hours=2),
         # 'execution_timeout': timedelta(seconds=300),
         # 'on_failure_callback': some_function, # or list of functions
         # 'on_success_callback': some_other_function, # or list of functions
diff --git a/airflow/models/__init__.py b/airflow/models/__init__.py
index 375761bc20f52..1a998d60c5c92 100644
--- a/airflow/models/__init__.py
+++ b/airflow/models/__init__.py
@@ -41,7 +41,6 @@
     "Pool",
     "RenderedTaskInstanceFields",
     "SkipMixin",
-    "SlaMiss",
     "TaskFail",
     "TaskInstance",
     "TaskReschedule",
@@ -104,7 +103,6 @@ def __getattr__(name):
     "Pool": "airflow.models.pool",
     "RenderedTaskInstanceFields": "airflow.models.renderedtifields",
     "SkipMixin": "airflow.models.skipmixin",
-    "SlaMiss": "airflow.models.slamiss",
     "TaskFail": "airflow.models.taskfail",
     "TaskInstance": "airflow.models.taskinstance",
     "TaskReschedule": "airflow.models.taskreschedule",
@@ -134,7 +132,6 @@ def __getattr__(name):
     from airflow.models.pool import Pool
     from airflow.models.renderedtifields import RenderedTaskInstanceFields
     from airflow.models.skipmixin import SkipMixin
-    from airflow.models.slamiss import SlaMiss
     from airflow.models.taskfail import TaskFail
     from airflow.models.taskinstance import TaskInstance, clear_task_instances
     from airflow.models.taskinstancehistory import TaskInstanceHistory
diff --git a/airflow/models/slamiss.py b/airflow/models/slamiss.py
deleted file mode 100644
index 4fb7e53a17bab..0000000000000
--- a/airflow/models/slamiss.py
+++ /dev/null
@@ -1,46 +0,0 @@
-#
-# 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.
-from __future__ import annotations
-
-from sqlalchemy import Boolean, Column, Index, String, Text
-
-from airflow.models.base import COLLATION_ARGS, ID_LEN, Base
-from airflow.utils.sqlalchemy import UtcDateTime
-
-
-class SlaMiss(Base):
-    """
-    Model that stores a history of the SLA that have been missed.
-
-    It is used to keep track of SLA failures over time and to avoid double triggering alert emails.
-    """
-
-    __tablename__ = "sla_miss"
-
-    task_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
-    dag_id = Column(String(ID_LEN, **COLLATION_ARGS), primary_key=True)
-    execution_date = Column(UtcDateTime, primary_key=True)
-    email_sent = Column(Boolean, default=False)
-    timestamp = Column(UtcDateTime)
-    description = Column(Text)
-    notification_sent = Column(Boolean, default=False)
-
-    __table_args__ = (Index("sm_dag", dag_id, unique=False),)
-
-    def __repr__(self):
-        return str((self.dag_id, self.task_id, self.execution_date.isoformat()))
diff --git a/airflow/notifications/basenotifier.py b/airflow/notifications/basenotifier.py
index 91fef87167ccb..8fbd339eee332 100644
--- a/airflow/notifications/basenotifier.py
+++ b/airflow/notifications/basenotifier.py
@@ -20,6 +20,7 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Sequence
 
+from airflow.exceptions import AirflowException
 from airflow.template.templater import Templater
 from airflow.utils.context import context_merge
 
@@ -87,10 +88,6 @@ def __call__(self, *args) -> None:
 
         :param context: The airflow context
         """
-        # Currently, there are two ways a callback is invoked
-        # 1. callback(context) - for on_*_callbacks
-        # 2. callback(dag, task_list, blocking_task_list, slas, blocking_tis) - for sla_miss_callback
-        # we have to distinguish between the two calls so that we can prepare the correct context,
         if len(args) == 1:
             context = args[0]
         else:
@@ -98,9 +95,9 @@ def __call__(self, *args) -> None:
                 "dag": args[0],
                 "task_list": args[1],
                 "blocking_task_list": args[2],
-                "slas": args[3],
-                "blocking_tis": args[4],
+                "blocking_tis": args[3],
             }
+
         self._update_context(context)
         self.render_template_fields(context)
         try:
diff --git a/airflow/utils/context.py b/airflow/utils/context.py
index e5d30b1e2d7d2..31f7daa8b97af 100644
--- a/airflow/utils/context.py
+++ b/airflow/utils/context.py
@@ -418,9 +418,13 @@ def context_merge(context: Context, *args: Any, **kwargs: Any) -> None:
 
     :meta private:
     """
+    if not context:
+        context = {}
+
     context.update(*args, **kwargs)
 
 
+
 def context_update_for_unmapped(context: Context, task: BaseOperator) -> None:
     """
     Update context after task unmapping.
diff --git a/airflow/www/auth.py b/airflow/www/auth.py
index 74f31d135c1aa..d4b8ad619e6e5 100644
--- a/airflow/www/auth.py
+++ b/airflow/www/auth.py
@@ -49,7 +49,7 @@
         IsAuthorizedPoolRequest,
         IsAuthorizedVariableRequest,
     )
-    from airflow.models import DagRun, Pool, SlaMiss, TaskInstance, Variable
+    from airflow.models import DagRun, Pool, TaskInstance, Variable
     from airflow.models.connection import Connection
     from airflow.models.xcom import BaseXCom
 
@@ -239,7 +239,7 @@ def has_access_dag_entities(method: ResourceMethod, access_entity: DagAccessEnti
     def has_access_decorator(func: T):
         @wraps(func)
         def decorated(*args, **kwargs):
-            items: set[SlaMiss | BaseXCom | DagRun | TaskInstance] = set(args[1])
+            items: set[BaseXCom | DagRun | TaskInstance] = set(args[1])
             requests: Sequence[IsAuthorizedDagRequest] = [
                 {
                     "method": method,
diff --git a/airflow/www/extensions/init_views.py b/airflow/www/extensions/init_views.py
index a04116a7c7d4d..1b0eddf55420a 100644
--- a/airflow/www/extensions/init_views.py
+++ b/airflow/www/extensions/init_views.py
@@ -103,9 +103,6 @@ def init_appbuilder_views(app):
     appbuilder.add_view(
         views.ConnectionModelView, permissions.RESOURCE_CONNECTION, category=permissions.RESOURCE_ADMIN_MENU
     )
-    appbuilder.add_view(
-        views.SlaMissModelView, permissions.RESOURCE_SLA_MISS, category=permissions.RESOURCE_BROWSE_MENU
-    )
     appbuilder.add_view(
         views.PluginView, permissions.RESOURCE_PLUGIN, category=permissions.RESOURCE_ADMIN_MENU
     )
diff --git a/airflow/www/views.py b/airflow/www/views.py
index 47c548d5e7667..2f1f397e51e28 100644
--- a/airflow/www/views.py
+++ b/airflow/www/views.py
@@ -103,7 +103,7 @@
 from airflow.jobs.job import Job
 from airflow.jobs.scheduler_job_runner import SchedulerJobRunner
 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
-from airflow.models import Connection, DagModel, DagTag, Log, SlaMiss, Trigger, XCom
+from airflow.models import Connection, DagModel, DagTag, Log, Trigger, XCom
 from airflow.models.asset import AssetDagRunQueue, AssetEvent, AssetModel, DagScheduleAssetReference
 from airflow.models.dag import get_asset_triggered_next_run_info
 from airflow.models.dagrun import RUN_ID_REGEX, DagRun, DagRunType
@@ -3876,119 +3876,6 @@ def action_post(self):
         return action.func(items)
 
 
-class SlaMissModelView(AirflowModelView):
-    """View to show SlaMiss table."""
-
-    route_base = "/slamiss"
-
-    datamodel = AirflowModelView.CustomSQLAInterface(SlaMiss)  # type: ignore
-
-    class_permission_name = permissions.RESOURCE_SLA_MISS
-    method_permission_name = {
-        "list": "read",
-        "action_muldelete": "delete",
-        "action_mulnotificationsent": "edit",
-        "action_mulnotificationsentfalse": "edit",
-        "action_mulemailsent": "edit",
-        "action_mulemailsentfalse": "edit",
-    }
-
-    base_permissions = [
-        permissions.ACTION_CAN_READ,
-        permissions.ACTION_CAN_ACCESS_MENU,
-    ]
-
-    list_columns = ["dag_id", "task_id", "execution_date", "email_sent", "notification_sent", "timestamp"]
-
-    label_columns = {
-        "execution_date": "Logical Date",
-    }
-
-    add_columns = ["dag_id", "task_id", "execution_date", "email_sent", "notification_sent", "timestamp"]
-    edit_columns = ["dag_id", "task_id", "execution_date", "email_sent", "notification_sent", "timestamp"]
-    search_columns = ["dag_id", "task_id", "email_sent", "notification_sent", "timestamp", "execution_date"]
-    base_order = ("execution_date", "desc")
-    base_filters = [["dag_id", DagFilter, list]]
-
-    formatters_columns = {
-        "task_id": wwwutils.task_instance_link,
-        "execution_date": wwwutils.datetime_f("execution_date"),
-        "timestamp": wwwutils.datetime_f("timestamp"),
-        "dag_id": wwwutils.dag_link,
-        "map_index": wwwutils.format_map_index,
-    }
-
-    @action("muldelete", "Delete", "Are you sure you want to delete selected records?", single=False)
-    @auth.has_access_dag_entities("DELETE", DagAccessEntity.SLA_MISS)
-    def action_muldelete(self, items):
-        """Multiple delete action."""
-        self.datamodel.delete_all(items)
-        self.update_redirect()
-        return redirect(self.get_redirect())
-
-    @action(
-        "mulnotificationsent",
-        "Set notification sent to true",
-        "Are you sure you want to set all these notifications to sent?",
-        single=False,
-    )
-    @auth.has_access_dag_entities("PUT", DagAccessEntity.SLA_MISS)
-    def action_mulnotificationsent(self, items: list[SlaMiss]):
-        return self._set_notification_property(items, "notification_sent", True)
-
-    @action(
-        "mulnotificationsentfalse",
-        "Set notification sent to false",
-        "Are you sure you want to mark these SLA alerts as notification not sent yet?",
-        single=False,
-    )
-    @auth.has_access_dag_entities("PUT", DagAccessEntity.SLA_MISS)
-    def action_mulnotificationsentfalse(self, items: list[SlaMiss]):
-        return self._set_notification_property(items, "notification_sent", False)
-
-    @action(
-        "mulemailsent",
-        "Set email sent to true",
-        "Are you sure you want to mark these SLA alerts as emails were sent?",
-        single=False,
-    )
-    @auth.has_access_dag_entities("PUT", DagAccessEntity.SLA_MISS)
-    def action_mulemailsent(self, items: list[SlaMiss]):
-        return self._set_notification_property(items, "email_sent", True)
-
-    @action(
-        "mulemailsentfalse",
-        "Set email sent to false",
-        "Are you sure you want to mark these SLA alerts as emails not sent yet?",
-        single=False,
-    )
-    @auth.has_access_dag_entities("PUT", DagAccessEntity.SLA_MISS)
-    def action_mulemailsentfalse(self, items: list[SlaMiss]):
-        return self._set_notification_property(items, "email_sent", False)
-
-    @provide_session
-    def _set_notification_property(
-        self,
-        items: list[SlaMiss],
-        attr: str,
-        new_value: bool,
-        session: Session = NEW_SESSION,
-    ):
-        try:
-            count = 0
-            for sla in items:
-                count += 1
-                setattr(sla, attr, new_value)
-                session.merge(sla)
-            session.commit()
-            flash(f"{count} SLAMisses had {attr} set to {new_value}.")
-        except Exception as ex:
-            flash(str(ex), "error")
-            flash("Failed to set state", "error")
-        self.update_redirect()
-        return redirect(self.get_default_url())
-
-
 class XComModelView(AirflowModelView):
     """View to show records from XCom table."""
 
diff --git a/tests/api_connexion/endpoints/test_mapped_task_instance_endpoint.py b/tests/api_connexion/endpoints/test_mapped_task_instance_endpoint.py
index fc53b8952f4aa..3615c1b2a5b70 100644
--- a/tests/api_connexion/endpoints/test_mapped_task_instance_endpoint.py
+++ b/tests/api_connexion/endpoints/test_mapped_task_instance_endpoint.py
@@ -33,7 +33,7 @@
 from airflow.utils.state import State, TaskInstanceState
 from airflow.utils.timezone import datetime
 from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user
-from tests.test_utils.db import clear_db_runs, clear_db_sla_miss, clear_rendered_ti_fields
+from tests.test_utils.db import clear_db_runs, clear_rendered_ti_fields
 from tests.test_utils.mock_operators import MockOperator
 
 pytestmark = [pytest.mark.db_test, pytest.mark.skip_if_database_isolation_mode]
@@ -81,7 +81,6 @@ def setup_attrs(self, configured_app) -> None:
         self.app = configured_app
         self.client = self.app.test_client()  # type:ignore
         clear_db_runs()
-        clear_db_sla_miss()
         clear_rendered_ti_fields()
 
     def create_dag_runs_with_mapped_tasks(self, dag_maker, session, dags=None):
@@ -227,7 +226,6 @@ def test_mapped_task_instances(self, one_task_with_mapped_tis, session):
             "queued_when": None,
             "rendered_fields": {},
             "rendered_map_index": None,
-            "sla_miss": None,
             "start_date": "2020-01-01T00:00:00+00:00",
             "state": "success",
             "task_id": "task_2",
diff --git a/tests/api_connexion/endpoints/test_task_instance_endpoint.py b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
index b5b3163e988d0..be92aa9a0dba1 100644
--- a/tests/api_connexion/endpoints/test_task_instance_endpoint.py
+++ b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
@@ -27,7 +27,7 @@
 
 from airflow.jobs.job import Job
 from airflow.jobs.triggerer_job_runner import TriggererJobRunner
-from airflow.models import DagRun, SlaMiss, TaskInstance, Trigger
+from airflow.models import DagRun, TaskInstance, Trigger
 from airflow.models.renderedtifields import RenderedTaskInstanceFields as RTIF
 from airflow.models.taskinstancehistory import TaskInstanceHistory
 from airflow.utils.platform import getuser
@@ -36,7 +36,7 @@
 from airflow.utils.timezone import datetime
 from airflow.utils.types import DagRunType
 from tests.test_utils.api_connexion_utils import assert_401, create_user, delete_user
-from tests.test_utils.db import clear_db_runs, clear_db_sla_miss, clear_rendered_ti_fields
+from tests.test_utils.db import clear_db_runs, clear_rendered_ti_fields
 from tests.test_utils.www import _check_last_log
 
 pytestmark = [pytest.mark.db_test, pytest.mark.skip_if_database_isolation_mode]
@@ -85,7 +85,6 @@ def setup_attrs(self, configured_app, dagbag) -> None:
         self.app = configured_app
         self.client = self.app.test_client()  # type:ignore
         clear_db_runs()
-        clear_db_sla_miss()
         clear_rendered_ti_fields()
         self.dagbag = dagbag
 
@@ -197,7 +196,6 @@ def test_should_respond_200(self, session):
             "priority_weight": 9,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": None,
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "running",
             "task_id": "print_the_context",
@@ -256,7 +254,6 @@ def test_should_respond_200_with_task_state_in_deferred(self, session):
             "priority_weight": 9,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": None,
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "deferred",
             "task_id": "print_the_context",
@@ -304,7 +301,6 @@ def test_should_respond_200_with_task_state_in_removed(self, session):
             "priority_weight": 9,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": None,
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "removed",
             "task_id": "print_the_context",
@@ -318,16 +314,9 @@ def test_should_respond_200_with_task_state_in_removed(self, session):
             "triggerer_job": None,
         }
 
-    def test_should_respond_200_task_instance_with_sla_and_rendered(self, session):
+    def test_should_respond_200_task_instance_with_rendered(self, session):
         tis = self.create_task_instances(session)
         session.query()
-        sla_miss = SlaMiss(
-            task_id="print_the_context",
-            dag_id="example_python_operator",
-            execution_date=self.default_time,
-            timestamp=self.default_time,
-        )
-        session.add(sla_miss)
         rendered_fields = RTIF(tis[0], render_templates=False)
         session.add(rendered_fields)
         session.commit()
@@ -355,15 +344,6 @@ def test_should_respond_200_task_instance_with_sla_and_rendered(self, session):
             "priority_weight": 9,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": {
-                "dag_id": "example_python_operator",
-                "description": None,
-                "email_sent": False,
-                "execution_date": "2020-01-01T00:00:00+00:00",
-                "notification_sent": False,
-                "task_id": "print_the_context",
-                "timestamp": "2020-01-01T00:00:00+00:00",
-            },
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "running",
             "task_id": "print_the_context",
@@ -416,7 +396,6 @@ def test_should_respond_200_mapped_task_instance_with_rtif(self, session):
                 "priority_weight": 9,
                 "queue": "default_queue",
                 "queued_when": None,
-                "sla_miss": None,
                 "start_date": "2020-01-02T00:00:00+00:00",
                 "state": "running",
                 "task_id": "print_the_context",
@@ -2320,7 +2299,6 @@ def test_should_respond_200(self, session):
             "priority_weight": 9,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": None,
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "running",
             "task_id": "print_the_context",
@@ -2380,7 +2358,6 @@ def test_should_respond_200_mapped_task_instance_with_rtif(self, session):
                 "priority_weight": 9,
                 "queue": "default_queue",
                 "queued_when": None,
-                "sla_miss": None,
                 "start_date": "2020-01-02T00:00:00+00:00",
                 "state": "running",
                 "task_id": "print_the_context",
diff --git a/tests/api_connexion/schemas/test_task_instance_schema.py b/tests/api_connexion/schemas/test_task_instance_schema.py
index 181e8351d2f75..4af699981b30b 100644
--- a/tests/api_connexion/schemas/test_task_instance_schema.py
+++ b/tests/api_connexion/schemas/test_task_instance_schema.py
@@ -26,7 +26,7 @@
     set_task_instance_state_form,
     task_instance_schema,
 )
-from airflow.models import RenderedTaskInstanceFields as RTIF, SlaMiss, TaskInstance as TI
+from airflow.models import RenderedTaskInstanceFields as RTIF, TaskInstance as TI
 from airflow.operators.empty import EmptyOperator
 from airflow.utils.platform import getuser
 from airflow.utils.state import State
@@ -64,7 +64,7 @@ def set_attrs(self, session, dag_maker):
 
         session.rollback()
 
-    def test_task_instance_schema_without_sla_and_rendered(self, session):
+    def test_task_instance_schema_without_rendered(self, session):
         ti = TI(task=self.task, **self.default_ti_init)
         session.add(ti)
         for key, value in self.default_ti_extras.items():
@@ -88,7 +88,6 @@ def test_task_instance_schema_without_sla_and_rendered(self, session):
             "priority_weight": 1,
             "queue": "default_queue",
             "queued_when": None,
-            "sla_miss": None,
             "start_date": "2020-01-02T00:00:00+00:00",
             "state": "running",
             "task_id": "TEST_TASK_ID",
@@ -103,63 +102,6 @@ def test_task_instance_schema_without_sla_and_rendered(self, session):
         }
         assert serialized_ti == expected_json
 
-    def test_task_instance_schema_with_sla_and_rendered(self, session):
-        sla_miss = SlaMiss(
-            task_id="TEST_TASK_ID",
-            dag_id="TEST_DAG_ID",
-            execution_date=self.default_time,
-        )
-        session.add(sla_miss)
-        session.flush()
-        ti = TI(task=self.task, **self.default_ti_init)
-        session.add(ti)
-        for key, value in self.default_ti_extras.items():
-            setattr(ti, key, value)
-        self.task.template_fields = ["partitions"]
-        setattr(self.task, "partitions", "data/ds=2022-02-17")
-        ti.rendered_task_instance_fields = RTIF(ti, render_templates=False)
-        serialized_ti = task_instance_schema.dump((ti, sla_miss))
-        expected_json = {
-            "dag_id": "TEST_DAG_ID",
-            "duration": 10000.0,
-            "end_date": "2020-01-03T00:00:00+00:00",
-            "execution_date": "2020-01-01T00:00:00+00:00",
-            "executor": None,
-            "executor_config": "{}",
-            "hostname": "",
-            "map_index": -1,
-            "max_tries": 0,
-            "note": "added some notes",
-            "operator": "EmptyOperator",
-            "pid": 100,
-            "pool": "default_pool",
-            "pool_slots": 1,
-            "priority_weight": 1,
-            "queue": "default_queue",
-            "queued_when": None,
-            "sla_miss": {
-                "dag_id": "TEST_DAG_ID",
-                "description": None,
-                "email_sent": False,
-                "execution_date": "2020-01-01T00:00:00+00:00",
-                "notification_sent": False,
-                "task_id": "TEST_TASK_ID",
-                "timestamp": None,
-            },
-            "start_date": "2020-01-02T00:00:00+00:00",
-            "state": "running",
-            "task_id": "TEST_TASK_ID",
-            "task_display_name": "TEST_TASK_ID",
-            "try_number": 0,
-            "unixname": getuser(),
-            "dag_run_id": None,
-            "rendered_fields": {"partitions": "data/ds=2022-02-17"},
-            "rendered_map_index": None,
-            "trigger": None,
-            "triggerer_job": None,
-        }
-        assert serialized_ti == expected_json
-
 
 class TestClearTaskInstanceFormSchema:
     @pytest.mark.parametrize(
diff --git a/tests/jobs/test_scheduler_job.py b/tests/jobs/test_scheduler_job.py
index b4e4c10cff456..a4138e017e143 100644
--- a/tests/jobs/test_scheduler_job.py
+++ b/tests/jobs/test_scheduler_job.py
@@ -84,7 +84,6 @@
     clear_db_pools,
     clear_db_runs,
     clear_db_serialized_dags,
-    clear_db_sla_miss,
     set_default_pool_slots,
 )
 from tests.test_utils.mock_executor import MockExecutor
@@ -142,7 +141,6 @@ def clean_db():
         clear_db_backfills()
         clear_db_pools()
         clear_db_dags()
-        clear_db_sla_miss()
         clear_db_import_errors()
         clear_db_jobs()
         clear_db_assets()
@@ -6321,7 +6319,6 @@ def clean_db():
         clear_db_pools()
         clear_db_backfills()
         clear_db_dags()
-        clear_db_sla_miss()
         clear_db_import_errors()
         clear_db_jobs()
         clear_db_serialized_dags()
diff --git a/tests/listeners/test_dag_import_error_listener.py b/tests/listeners/test_dag_import_error_listener.py
index 57fd4f79dd526..70c4e1e08ac91 100644
--- a/tests/listeners/test_dag_import_error_listener.py
+++ b/tests/listeners/test_dag_import_error_listener.py
@@ -40,7 +40,6 @@
     clear_db_pools,
     clear_db_runs,
     clear_db_serialized_dags,
-    clear_db_sla_miss,
 )
 from tests.test_utils.mock_executor import MockExecutor
 
@@ -75,7 +74,6 @@ def clean_db():
         clear_db_runs()
         clear_db_pools()
         clear_db_dags()
-        clear_db_sla_miss()
         clear_db_import_errors()
         clear_db_jobs()
         clear_db_serialized_dags()
diff --git a/tests/notifications/test_basenotifier.py b/tests/notifications/test_basenotifier.py
index 65bda6ade8ade..b2e3d751a6f30 100644
--- a/tests/notifications/test_basenotifier.py
+++ b/tests/notifications/test_basenotifier.py
@@ -94,7 +94,6 @@ def test_notifier_call_with_prepared_context(self, dag_maker, caplog):
                 "dag": None,
                 "task_list": ["some_task"],
                 "blocking_task_list": None,
-                "slas": None,
                 "blocking_tis": None,
                 "message": "task: {{ task_list[0] }}",
             }
diff --git a/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py b/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
index 69b3c221eae93..72667bd343c26 100644
--- a/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
+++ b/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
@@ -34,7 +34,7 @@
     delete_user,
 )
 from tests.test_utils.compat import AIRFLOW_V_3_0_PLUS
-from tests.test_utils.db import clear_db_runs, clear_db_sla_miss, clear_rendered_ti_fields
+from tests.test_utils.db import clear_db_runs, clear_rendered_ti_fields
 
 pytestmark = [
     pytest.mark.db_test,
@@ -123,7 +123,6 @@ def setup_attrs(self, configured_app, dagbag) -> None:
         self.app = configured_app
         self.client = self.app.test_client()  # type:ignore
         clear_db_runs()
-        clear_db_sla_miss()
         clear_rendered_ti_fields()
         self.dagbag = dagbag
 
diff --git a/tests/providers/smtp/notifications/test_smtp.py b/tests/providers/smtp/notifications/test_smtp.py
index 39b51e8e02ce7..351a24c449dc7 100644
--- a/tests/providers/smtp/notifications/test_smtp.py
+++ b/tests/providers/smtp/notifications/test_smtp.py
@@ -23,7 +23,6 @@
 import pytest
 
 from airflow.configuration import conf
-from airflow.models import SlaMiss
 from airflow.operators.empty import EmptyOperator
 from airflow.providers.smtp.hooks.smtp import SmtpHook
 from airflow.providers.smtp.notifications.smtp import (
@@ -145,38 +144,6 @@ def test_notifier_with_defaults(self, mock_smtphook_hook, create_task_instance):
         content = mock_smtphook_hook.return_value.__enter__().send_email_smtp.call_args.kwargs["html_content"]
         assert f"{NUM_TRY} of 1" in content
 
-    @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
-    def test_notifier_with_defaults_sla(self, mock_smtphook_hook, dag_maker):
-        with dag_maker("test_notifier") as dag:
-            EmptyOperator(task_id="task1")
-        context = {
-            "dag": dag,
-            "slas": [SlaMiss(task_id="op", dag_id=dag.dag_id, execution_date=timezone.datetime(2018, 1, 1))],
-            "task_list": [],
-            "blocking_task_list": [],
-            "blocking_tis": [],
-        }
-        notifier = SmtpNotifier(
-            from_email=conf.get("smtp", "smtp_mail_from"),
-            to="test_reciver@test.com",
-        )
-        notifier(context)
-        mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
-            from_email=conf.get("smtp", "smtp_mail_from"),
-            to="test_reciver@test.com",
-            subject="SLA Missed for DAG test_notifier - Task op",
-            html_content=mock.ANY,
-            smtp_conn_id="smtp_default",
-            files=None,
-            cc=None,
-            bcc=None,
-            mime_subtype="mixed",
-            mime_charset="utf-8",
-            custom_headers=None,
-        )
-        content = mock_smtphook_hook.return_value.__enter__().send_email_smtp.call_args.kwargs["html_content"]
-        assert "Task List:" in content
-
     @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
     def test_notifier_with_nondefault_conf_vars(self, mock_smtphook_hook, create_task_instance):
         ti = create_task_instance(dag_id="dag", task_id="op", execution_date=timezone.datetime(2018, 1, 1))
diff --git a/tests/test_utils/db.py b/tests/test_utils/db.py
index a5dd94e2d009d..f4992620210ec 100644
--- a/tests/test_utils/db.py
+++ b/tests/test_utils/db.py
@@ -27,7 +27,6 @@
     Log,
     Pool,
     RenderedTaskInstanceFields,
-    SlaMiss,
     TaskFail,
     TaskInstance,
     TaskReschedule,
@@ -108,11 +107,6 @@ def clear_db_serialized_dags():
         session.query(SerializedDagModel).delete()
 
 
-def clear_db_sla_miss():
-    with create_session() as session:
-        session.query(SlaMiss).delete()
-
-
 def clear_db_pools():
     with create_session() as session:
         session.query(Pool).delete()
@@ -235,7 +229,6 @@ def clear_all():
     clear_db_assets()
     clear_db_dags()
     clear_db_serialized_dags()
-    clear_db_sla_miss()
     clear_db_dag_code()
     clear_db_callbacks()
     clear_rendered_ti_fields()

From 105ff30dd9921c218eeb6eb2189eea5e58fced16 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Tue, 8 Oct 2024 16:05:35 -0700
Subject: [PATCH 2/8] Untuple the TI dump

---
 .../endpoints/task_instance_endpoint.py            |  8 +++-----
 .../api_connexion/schemas/task_instance_schema.py  | 14 +++++---------
 2 files changed, 8 insertions(+), 14 deletions(-)

diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index 7980012488e18..3d7d6438baaee 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -639,17 +639,15 @@ def set_task_instance_note(
         query = query.where(TI.map_index == map_index)
 
     try:
-        result = session.execute(query).one_or_none()
+        ti = session.scalar(query)
     except MultipleResultsFound:
         raise NotFound(
             "Task instance not found", detail="Task instance is mapped, add the map_index value to the URL"
         )
-    if result is None:
+    if ti is None:
         error_message = f"Task Instance not found for dag_id={dag_id}, run_id={dag_run_id}, task_id={task_id}"
         raise NotFound(error_message)
 
-    ti = result[0]
-
     current_user_id = get_auth_manager().get_user_id()
     if ti.task_instance_note is None:
         ti.note = (new_note, current_user_id)
@@ -657,7 +655,7 @@ def set_task_instance_note(
         ti.task_instance_note.content = new_note
         ti.task_instance_note.user_id = current_user_id
     session.commit()
-    return task_instance_schema.dump((ti,))
+    return task_instance_schema.dump(ti)
 
 
 @security.requires_access_dag("GET", DagAccessEntity.TASK_INSTANCE)
diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 08b7cfdbac9cb..eace461278b36 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -74,15 +74,11 @@ class Meta:
     triggerer_job = fields.Nested(JobSchema)
 
     def get_attribute(self, obj, attr, default):
-        if attr == "sla_miss":
-            # Object is a tuple of task_instance and slamiss
-            # and the get_value expects a dict with key, value
-            # corresponding to the attr.
-            slamiss_instance = {"sla_miss": obj[1]}
-            return get_value(slamiss_instance, attr, default)
-        elif attr == "rendered_fields":
-            return get_value(obj[0], "rendered_task_instance_fields.rendered_fields", default)
-        return get_value(obj[0], attr, default)
+        ti = obj if isinstance(obj, TaskInstance) else obj[0]
+
+        if attr == "rendered_fields":
+            return get_value(ti, "rendered_task_instance_fields.rendered_fields", default)
+        return get_value(ti, attr, default)
 
 
 class TaskInstanceHistorySchema(SQLAlchemySchema):

From ec9e93e81b2210ba8d3b035bb34d15086edcf544 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Wed, 9 Oct 2024 10:20:31 -0700
Subject: [PATCH 3/8] A little more cleaning

---
 tests/api_connexion/schemas/test_task_instance_schema.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/api_connexion/schemas/test_task_instance_schema.py b/tests/api_connexion/schemas/test_task_instance_schema.py
index 4af699981b30b..5c9348ddeaa4c 100644
--- a/tests/api_connexion/schemas/test_task_instance_schema.py
+++ b/tests/api_connexion/schemas/test_task_instance_schema.py
@@ -69,7 +69,7 @@ def test_task_instance_schema_without_rendered(self, session):
         session.add(ti)
         for key, value in self.default_ti_extras.items():
             setattr(ti, key, value)
-        serialized_ti = task_instance_schema.dump((ti, None, None))
+        serialized_ti = task_instance_schema.dump(ti)
         expected_json = {
             "dag_id": "TEST_DAG_ID",
             "duration": 10000.0,

From 2ea468f41190ac5745b84da2818862b4e01c2d6f Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Wed, 9 Oct 2024 16:03:11 -0700
Subject: [PATCH 4/8] fix sqlalchemy

Co-Authored-By: dstandish <dstandish@users.noreply.github.com>
---
 .../api_connexion/endpoints/task_instance_endpoint.py  | 10 +++++-----
 airflow/api_connexion/schemas/task_instance_schema.py  |  6 ++----
 2 files changed, 7 insertions(+), 9 deletions(-)

diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index 3d7d6438baaee..3e1d8ca384dcd 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -87,14 +87,14 @@ def get_task_instance(
     )
 
     try:
-        task_instance = session.execute(query).one_or_none()
+        task_instance = session.scalar(query)
     except MultipleResultsFound:
         raise NotFound(
             "Task instance not found", detail="Task instance is mapped, add the map_index value to the URL"
         )
     if task_instance is None:
         raise NotFound("Task instance not found")
-    if task_instance[0].map_index != -1:
+    if task_instance.map_index != -1:
         raise NotFound(
             "Task instance not found", detail="Task instance is mapped, add the map_index value to the URL"
         )
@@ -119,7 +119,7 @@ def get_mapped_task_instance(
         .join(TI.dag_run)
         .options(joinedload(TI.rendered_task_instance_fields))
     )
-    task_instance = session.execute(query).one_or_none()
+    task_instance = session.scalar(query)
 
     if task_instance is None:
         raise NotFound("Task instance not found")
@@ -356,7 +356,7 @@ def get_task_instances(
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
+    task_instances = session.scalars(entry_query.offset(offset).limit(limit))
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
     )
@@ -425,7 +425,7 @@ def get_task_instances_batch(session: Session = NEW_SESSION) -> APIResponse:
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    task_instances = session.execute(ti_query).all()
+    task_instances = session.scalars(ti_query)
 
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index eace461278b36..75eab729635df 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -74,11 +74,9 @@ class Meta:
     triggerer_job = fields.Nested(JobSchema)
 
     def get_attribute(self, obj, attr, default):
-        ti = obj if isinstance(obj, TaskInstance) else obj[0]
-
         if attr == "rendered_fields":
-            return get_value(ti, "rendered_task_instance_fields.rendered_fields", default)
-        return get_value(ti, attr, default)
+            return get_value(obj, "rendered_task_instance_fields.rendered_fields", default)
+        return get_value(obj, attr, default)
 
 
 class TaskInstanceHistorySchema(SQLAlchemySchema):

From 002ae49e3a5cdd68e09c9c9731364d6d75c5b4f3 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 11 Oct 2024 12:07:31 -0700
Subject: [PATCH 5/8] fix merge conflict mistakes

---
 airflow/api_connexion/endpoints/task_instance_endpoint.py | 2 +-
 airflow/notifications/basenotifier.py                     | 1 -
 airflow/utils/context.py                                  | 1 -
 dev/tests_common/test_utils/db.py                         | 6 ------
 tests/api_connexion/schemas/test_task_instance_schema.py  | 2 +-
 5 files changed, 2 insertions(+), 10 deletions(-)

diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index 3e1d8ca384dcd..da19d2ec4e60f 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -20,7 +20,7 @@
 
 from flask import g
 from marshmallow import ValidationError
-from sqlalchemy import and_, or_, select
+from sqlalchemy import or_, select
 from sqlalchemy.exc import MultipleResultsFound
 from sqlalchemy.orm import joinedload
 
diff --git a/airflow/notifications/basenotifier.py b/airflow/notifications/basenotifier.py
index 8fbd339eee332..11d7ee17fe939 100644
--- a/airflow/notifications/basenotifier.py
+++ b/airflow/notifications/basenotifier.py
@@ -20,7 +20,6 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Sequence
 
-from airflow.exceptions import AirflowException
 from airflow.template.templater import Templater
 from airflow.utils.context import context_merge
 
diff --git a/airflow/utils/context.py b/airflow/utils/context.py
index 31f7daa8b97af..f2ee0f9b47def 100644
--- a/airflow/utils/context.py
+++ b/airflow/utils/context.py
@@ -424,7 +424,6 @@ def context_merge(context: Context, *args: Any, **kwargs: Any) -> None:
     context.update(*args, **kwargs)
 
 
-
 def context_update_for_unmapped(context: Context, task: BaseOperator) -> None:
     """
     Update context after task unmapping.
diff --git a/dev/tests_common/test_utils/db.py b/dev/tests_common/test_utils/db.py
index c0cfff1110539..00e5cb5b10e63 100644
--- a/dev/tests_common/test_utils/db.py
+++ b/dev/tests_common/test_utils/db.py
@@ -130,11 +130,6 @@ def clear_db_serialized_dags():
         session.query(SerializedDagModel).delete()
 
 
-def clear_db_sla_miss():
-    with create_session() as session:
-        session.query(SlaMiss).delete()
-
-
 def clear_db_pools():
     with create_session() as session:
         session.query(Pool).delete()
@@ -257,7 +252,6 @@ def clear_all():
     clear_db_assets()
     clear_db_dags()
     clear_db_serialized_dags()
-    clear_db_sla_miss()
     clear_db_dag_code()
     clear_db_callbacks()
     clear_rendered_ti_fields()
diff --git a/tests/api_connexion/schemas/test_task_instance_schema.py b/tests/api_connexion/schemas/test_task_instance_schema.py
index 5c9348ddeaa4c..ff18ad75b5448 100644
--- a/tests/api_connexion/schemas/test_task_instance_schema.py
+++ b/tests/api_connexion/schemas/test_task_instance_schema.py
@@ -26,7 +26,7 @@
     set_task_instance_state_form,
     task_instance_schema,
 )
-from airflow.models import RenderedTaskInstanceFields as RTIF, TaskInstance as TI
+from airflow.models import TaskInstance as TI
 from airflow.operators.empty import EmptyOperator
 from airflow.utils.platform import getuser
 from airflow.utils.state import State

From fb446bb1ac86659f7e6a255057c728348e03deb8 Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 11 Oct 2024 12:20:40 -0700
Subject: [PATCH 6/8] more merge conflict fixes

---
 .../fab/auth_manager/api_endpoints/test_task_instance_endpoint.py | 0
 tests/providers/smtp/notifications/test_smtp.py                   | 0
 tests/test_utils/db.py                                            | 0
 3 files changed, 0 insertions(+), 0 deletions(-)
 delete mode 100644 tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
 delete mode 100644 tests/providers/smtp/notifications/test_smtp.py
 delete mode 100644 tests/test_utils/db.py

diff --git a/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py b/tests/providers/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
deleted file mode 100644
index e69de29bb2d1d..0000000000000
diff --git a/tests/providers/smtp/notifications/test_smtp.py b/tests/providers/smtp/notifications/test_smtp.py
deleted file mode 100644
index e69de29bb2d1d..0000000000000
diff --git a/tests/test_utils/db.py b/tests/test_utils/db.py
deleted file mode 100644
index e69de29bb2d1d..0000000000000

From 6cc27135fbaa9f1df08e05d964f48f9902c8487a Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 11 Oct 2024 14:33:17 -0700
Subject: [PATCH 7/8] missed an sqalchemy fix

---
 airflow/api_connexion/endpoints/task_instance_endpoint.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index da19d2ec4e60f..0e98173f68a5f 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -219,7 +219,7 @@ def get_mapped_task_instances(
     except _UnsupportedOrderBy as e:
         raise BadRequest(detail=f"Ordering with {e.order_by!r} is not supported")
 
-    task_instances = session.execute(entry_query.offset(offset).limit(limit)).all()
+    task_instances = session.scalars(entry_query.offset(offset).limit(limit))
     return task_instance_collection_schema.dump(
         TaskInstanceCollection(task_instances=task_instances, total_entries=total_entries)
     )

From ca5c913618d4df2b19f5c488f3a80cfc73ab21da Mon Sep 17 00:00:00 2001
From: ferruzzi <ferruzzi@amazon.com>
Date: Fri, 11 Oct 2024 14:34:22 -0700
Subject: [PATCH 8/8] I really messed up that merge...

---
 .../schemas/task_instance_schema.py           |  7 ++--
 airflow/utils/context.py                      |  2 +-
 .../test_task_instance_endpoint.py            |  3 +-
 .../tests/smtp/notifications/test_smtp.py     | 33 -------------------
 4 files changed, 4 insertions(+), 41 deletions(-)

diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 75eab729635df..f0b8285fdfa0e 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -16,7 +16,7 @@
 # under the License.
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, NamedTuple
+from typing import NamedTuple
 
 from marshmallow import Schema, ValidationError, fields, validate, validates_schema
 from marshmallow.utils import get_value
@@ -32,9 +32,6 @@
 from airflow.utils.helpers import exactly_one
 from airflow.utils.state import TaskInstanceState
 
-if TYPE_CHECKING:
-    from airflow.models import SlaMiss
-
 
 class TaskInstanceSchema(SQLAlchemySchema):
     """Task instance schema."""
@@ -114,7 +111,7 @@ class Meta:
 class TaskInstanceCollection(NamedTuple):
     """List of task instances with metadata."""
 
-    task_instances: list[tuple[TaskInstance, SlaMiss | None]]
+    task_instances: list[TaskInstance | None]
     total_entries: int
 
 
diff --git a/airflow/utils/context.py b/airflow/utils/context.py
index f2ee0f9b47def..4f308a6f9e060 100644
--- a/airflow/utils/context.py
+++ b/airflow/utils/context.py
@@ -419,7 +419,7 @@ def context_merge(context: Context, *args: Any, **kwargs: Any) -> None:
     :meta private:
     """
     if not context:
-        context = {}
+        context = Context()
 
     context.update(*args, **kwargs)
 
diff --git a/providers/tests/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py b/providers/tests/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
index 7d379b6f8c007..b806a003fd637 100644
--- a/providers/tests/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
+++ b/providers/tests/fab/auth_manager/api_endpoints/test_task_instance_endpoint.py
@@ -30,7 +30,7 @@
 from airflow.utils.types import DagRunType
 
 from dev.tests_common.test_utils.compat import AIRFLOW_V_3_0_PLUS
-from dev.tests_common.test_utils.db import clear_db_runs, clear_db_sla_miss, clear_rendered_ti_fields
+from dev.tests_common.test_utils.db import clear_db_runs, clear_rendered_ti_fields
 from providers.tests.fab.auth_manager.api_endpoints.api_connexion_utils import (
     create_user,
     delete_roles,
@@ -124,7 +124,6 @@ def setup_attrs(self, configured_app, dagbag) -> None:
         self.app = configured_app
         self.client = self.app.test_client()  # type:ignore
         clear_db_runs()
-        clear_db_sla_miss()
         clear_rendered_ti_fields()
         self.dagbag = dagbag
 
diff --git a/providers/tests/smtp/notifications/test_smtp.py b/providers/tests/smtp/notifications/test_smtp.py
index 75e1ebf530fae..5ee136099b7f0 100644
--- a/providers/tests/smtp/notifications/test_smtp.py
+++ b/providers/tests/smtp/notifications/test_smtp.py
@@ -23,7 +23,6 @@
 import pytest
 
 from airflow.configuration import conf
-from airflow.models import SlaMiss
 from airflow.operators.empty import EmptyOperator
 from airflow.providers.smtp.hooks.smtp import SmtpHook
 from airflow.providers.smtp.notifications.smtp import (
@@ -146,38 +145,6 @@ def test_notifier_with_defaults(self, mock_smtphook_hook, create_task_instance):
         content = mock_smtphook_hook.return_value.__enter__().send_email_smtp.call_args.kwargs["html_content"]
         assert f"{NUM_TRY} of 1" in content
 
-    @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
-    def test_notifier_with_defaults_sla(self, mock_smtphook_hook, dag_maker):
-        with dag_maker("test_notifier") as dag:
-            EmptyOperator(task_id="task1")
-        context = {
-            "dag": dag,
-            "slas": [SlaMiss(task_id="op", dag_id=dag.dag_id, execution_date=timezone.datetime(2018, 1, 1))],
-            "task_list": [],
-            "blocking_task_list": [],
-            "blocking_tis": [],
-        }
-        notifier = SmtpNotifier(
-            from_email=conf.get("smtp", "smtp_mail_from"),
-            to="test_reciver@test.com",
-        )
-        notifier(context)
-        mock_smtphook_hook.return_value.__enter__().send_email_smtp.assert_called_once_with(
-            from_email=conf.get("smtp", "smtp_mail_from"),
-            to="test_reciver@test.com",
-            subject="SLA Missed for DAG test_notifier - Task op",
-            html_content=mock.ANY,
-            smtp_conn_id="smtp_default",
-            files=None,
-            cc=None,
-            bcc=None,
-            mime_subtype="mixed",
-            mime_charset="utf-8",
-            custom_headers=None,
-        )
-        content = mock_smtphook_hook.return_value.__enter__().send_email_smtp.call_args.kwargs["html_content"]
-        assert "Task List:" in content
-
     @mock.patch("airflow.providers.smtp.notifications.smtp.SmtpHook")
     def test_notifier_with_nondefault_conf_vars(self, mock_smtphook_hook, create_task_instance):
         ti = create_task_instance(dag_id="dag", task_id="op", execution_date=timezone.datetime(2018, 1, 1))
