This method is used in DagFileProcessor.process_file - it may be easier to migrate all it's internal calls instead of the whole method.
continue
                tasks_missed_sla.append(task)
            emails: set[str] = set()
            for task in tasks_missed_sla:
                if task.email:
                    if isinstance(task.email, str):
                        emails |= set(get_email_address_list(task.email))
                    elif isinstance(task.email, (list, tuple)):
                        emails |= set(task.email)
            if emails:
                try:
                    send_email(emails, f"[airflow] SLA miss on DAG={dag.dag_id}", email_content)
                    email_sent = True
                    notification_sent = True
                except Exception:
                    Stats.incr("sla_email_notification_failure", tags={"dag_id": dag.dag_id})
                    cls.logger().exception(
                        "Could not send SLA Miss email notification for DAG %s", dag.dag_id
                    )
            
            if notification_sent:
                for sla in slas:
                    sla.email_sent = email_sent
                    sla.notification_sent = True
                    session.merge(sla)
            session.commit()
    @staticmethod
    @internal_api_call
    @provide_session
    def update_import_errors(