class_docstr: Runtime environment for all triggers.
  
  Mainly runs inside its own thread, where it hands control off to an asyncio
  event loop, but is also sometimes interacted with from the main thread
  (where all the DB queries are done). All communication between threads is
  done via Deques.
docstr: Request that we update what triggers we're running.
  
  Works out the differences - ones to add, and ones to remove - then
  adds them to the deques so the subthread can actually mutate the running
  trigger set.
docstr_tok: ["Request","that","we","update","what","triggers","we","'re","running",".","Works","out","the","differences","-","ones","to","add",",","and","ones","to","remove","-","then","adds","them","to","the","deques","so","the","subthread","can","actually","mutate","the","running","trigger","set","."]
code: def update_triggers(self, requested_trigger_ids: set[int]):
      """
      Request that we update what triggers we're running.
  
      Works out the differences - ones to add, and ones to remove - then
      adds them to the deques so the subthread can actually mutate the running
      trigger set.
      """
      # Note that `triggers` could be mutated by the other thread during this
      # line's execution, but we consider that safe, since there's a strict
      # add -> remove -> never again lifecycle this function is already
      # handling.
      running_trigger_ids = set(self.triggers.keys())
      known_trigger_ids = (
          running_trigger_ids.union(x[0] for x in self.events)
          .union(self.to_cancel)
          .union(x[0] for x in self.to_create)
          .union(trigger[0] for trigger in self.failed_triggers)
      )
      # Work out the two difference sets
      new_trigger_ids = requested_trigger_ids - known_trigger_ids
      cancel_trigger_ids = running_trigger_ids - requested_trigger_ids
      # Bulk-fetch new trigger records
      new_triggers = Trigger.bulk_fetch(new_trigger_ids)
      # Add in new triggers
      for new_id in new_trigger_ids:
          # Check it didn't vanish in the meantime
          if new_id not in new_triggers:
              self.log.warning(
                  "Trigger ID %s disappeared before we could start it",
                  new_id,
              )
              continue
          # Resolve trigger record into an actual class instance
          try:
              new_trigger_orm = new_triggers[new_id]
              trigger_class = self.get_trigger_by_classpath(
                  new_trigger_orm.classpath
              )
          except BaseException as e:
              # Either the trigger code or the path to it is bad. Fail the trigger.
              self.failed_triggers.append((new_id, e))
              continue
  
          try:
              new_trigger_instance = (
                  self.trigger_row_to_trigger_instance(
                      new_trigger_orm, trigger_class
                  )
              )
          except TypeError as err:
              self.log.error("Trigger failed; message=%s", err)
              self.failed_triggers.append((new_id, err))
              continue
  
          self.set_trigger_logging_metadata(
              new_trigger_orm.task_instance,
              new_id,
              new_trigger_instance,
          )
          self.to_create.append((new_id, new_trigger_instance))
      # Enqueue orphaned triggers for cancellation
      self.to_cancel.extend(cancel_trigger_ids)
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import os
  import signal
  import sys
  import threading
  import time
  import warnings
  from collections import deque
  from contextlib import suppress
  from copy import copy
  from queue import SimpleQueue
  from typing import TYPE_CHECKING, TypeVar
  from sqlalchemy import func, select
  from airflow.configuration import conf
  from airflow.jobs.base_job_runner import BaseJobRunner
  from airflow.jobs.job import perform_heartbeat
  from airflow.models.trigger import ENCRYPTED_KWARGS_PREFIX, Trigger
  from airflow.stats import Stats
  from airflow.triggers.base import BaseTrigger, TriggerEvent
  from airflow.typing_compat import TypedDict
  from airflow.utils import timezone
  from airflow.utils.log.file_task_handler import FileTaskHandler
  from airflow.utils.log.logging_mixin import LoggingMixin
  from airflow.utils.log.trigger_handler import DropTriggerLogsFilter, LocalQueueHandler, TriggererHandlerWrapper, TriggerMetadataFilter, ctx_indiv_trigger, ctx_task_instance, ctx_trigger_end, ctx_trigger_id
  from airflow.utils.module_loading import import_string
  from airflow.utils.session import NEW_SESSION, provide_session
label_desc: Use image node_id 5 for calling the TriggerRunner obj's underlying member method code with example usage: obj.update_triggers(requested_trigger_ids) without return types