class_docstr: Process a Python file containing Airflow DAGs.
  
  This includes:
  
  1. Execute the file and look for DAG objects in the namespace.
  2. Execute any Callbacks if passed to DagFileProcessor.process_file
  3. Serialize the DAGs and save it to DB (or update existing record in the DB).
  4. Pickle the DAG and save it to the DB (if necessary).
  5. Record any errors importing the file into ORM
  
  Returns a tuple of 'number of dags found' and 'the count of import errors'
  
  :param dag_ids: If specified, only look at these DAG ID's
  :param log: Logger to save the processing process
class_docstr_tok: ["Process","a","Python","file","containing","Airflow","DAGs",".","This","includes",":","1",".","Execute","the","file","and","look","for","DAG","objects","in","the","namespace",".","2",".","Execute","any","Callbacks","if","passed","to","DagFileProcessor.process_file","3",".","Serialize","the","DAGs","and","save","it","to","DB","(","or","update","existing","record","in","the","DB",")",".","4",".","Pickle","the","DAG","and","save","it","to","the","DB","(","if","necessary",")",".","5",".","Record","any","errors","importing","the","file","into","ORM","Returns","a","tuple","of","'number","of","dags","found","'","and","'the","count","of","import","errors'",":","param","dag_ids",":","If","specified",",","only","look","at","these","DAG","ID's",":","param","log",":","Logger","to","save","the","processing","process"]
docstr: Update any import errors to be displayed in the UI.
  
  For the DAGs in the given DagBag, record any associated import errors and clears
  errors for files that no longer have them. These are usually displayed through the
  Airflow UI so that users know that there are issues parsing DAGs.
  
  :param dagbag: DagBag containing DAGs with import errors
  :param session: session for ORM operations
docstr_tok: ["Update","any","import","errors","to","be","displayed","in","the","UI",".","For","the","DAGs","in","the","given","DagBag",",","record","any","associated","import","errors","and","clears","errors","for","files","that","no","longer","have","them",".","These","are","usually","displayed","through","the","Airflow","UI","so","that","users","know","that","there","are","issues","parsing","DAGs",".",":","param","dagbag",":","DagBag","containing","DAGs","with","import","errors",":","param","session",":","session","for","ORM","operations"]
code: def update_import_errors(
      file_last_changed: dict[str, datetime],
      import_errors: dict[str, str],
      processor_subdir: str | None,
      session: Session = NEW_SESSION,
  ) -> None:
      """
      Update any import errors to be displayed in the UI.
  
      For the DAGs in the given DagBag, record any associated import errors and clears
      errors for files that no longer have them. These are usually displayed through the
      Airflow UI so that users know that there are issues parsing DAGs.
  
      :param dagbag: DagBag containing DAGs with import errors
      :param session: session for ORM operations
      """
      files_without_error = file_last_changed - import_errors.keys()
  
      # Clear the errors of the processed files
      # that no longer have errors
      for dagbag_file in files_without_error:
          session.execute(
              delete(errors.ImportError)
              .where(
                  errors.ImportError.filename.startswith(dagbag_file)
              )
              .execution_options(synchronize_session="fetch")
          )
  
      # files that still have errors
      existing_import_error_files = [
          x.filename
          for x in session.query(errors.ImportError.filename).all()
      ]
  
      # Add the errors of the processed files
      for filename, stacktrace in import_errors.items():
          if filename in existing_import_error_files:
              session.query(errors.ImportError).filter(
                  errors.ImportError.filename == filename
              ).update(
                  {
                      "filename": filename,
                      "timestamp": timezone.utcnow(),
                      "stacktrace": stacktrace,
                  },
                  synchronize_session="fetch",
              )
          else:
              session.add(
                  errors.ImportError(
                      filename=filename,
                      timestamp=timezone.utcnow(),
                      stacktrace=stacktrace,
                      processor_subdir=processor_subdir,
                  )
              )
          (
              session.query(DagModel)
              .filter(DagModel.fileloc == filename)
              .update(
                  {"has_import_errors": True},
                  synchronize_session="fetch",
              )
          )
  
      session.commit()
  
imports: from __future__ import annotations
  import importlib
  import logging
  import os
  import signal
  import threading
  import time
  import zipfile
  from contextlib import redirect_stderr, redirect_stdout, suppress
  from datetime import timedelta
  from typing import TYPE_CHECKING, Iterable, Iterator
  from setproctitle import setproctitle
  from sqlalchemy import delete, func, or_, select
  from airflow import settings
  from airflow.api_internal.internal_api_call import internal_api_call
  from airflow.callbacks.callback_requests import DagCallbackRequest, SlaCallbackRequest, TaskCallbackRequest
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException, TaskNotFound
  from airflow.models import SlaMiss, errors
  from airflow.models.dag import DAG, DagModel
  from airflow.models.dagbag import DagBag
  from airflow.models.dagrun import DagRun
  from airflow.models.dagwarning import DagWarning, DagWarningType
  from airflow.models.serialized_dag import SerializedDagModel
  from airflow.models.taskinstance import TaskInstance, TaskInstance
  from airflow.stats import Stats
  from airflow.utils import timezone
  from airflow.utils.email import get_email_address_list, send_email
  from airflow.utils.file import iter_airflow_imports, might_contain_dag
  from airflow.utils.log.logging_mixin import LoggingMixin, StreamLogWriter, set_context
  from airflow.utils.mixins import MultiprocessingStartMethodMixin
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import TaskInstanceState
label_desc: Use image node_id 3 for calling the DagFileProcessor obj's underlying member method code with example usage: obj.update_import_errors(file_last_changed, import_errors, processor_subdir, session) without return types