code: def context_to_airflow_vars(
      context: Mapping[str, Any], in_env_var_format: bool = False
  ) -> dict[str, str]:
      """
      Return values used to externally reconstruct relations between dags, dag_runs, tasks and task_instances.
  
      Given a context, this function provides a dictionary of values that can be used to
      externally reconstruct relations between dags, dag_runs, tasks and task_instances.
      Default to abc.def.ghi format and can be made to ABC_DEF_GHI format if
      in_env_var_format is set to True.
  
      :param context: The context for the task_instance of interest.
      :param in_env_var_format: If returned vars should be in ABC_DEF_GHI format.
      :return: task_instance context as dict.
      """
      params = {}
      if in_env_var_format:
          name_format = "env_var_format"
      else:
          name_format = "default"
  
      task = context.get("task")
      task_instance = context.get("task_instance")
      dag_run = context.get("dag_run")
  
      ops = [
          (task, "email", "AIRFLOW_CONTEXT_DAG_EMAIL"),
          (task, "owner", "AIRFLOW_CONTEXT_DAG_OWNER"),
          (task_instance, "dag_id", "AIRFLOW_CONTEXT_DAG_ID"),
          (task_instance, "task_id", "AIRFLOW_CONTEXT_TASK_ID"),
          (
              task_instance,
              "execution_date",
              "AIRFLOW_CONTEXT_EXECUTION_DATE",
          ),
          (task_instance, "try_number", "AIRFLOW_CONTEXT_TRY_NUMBER"),
          (dag_run, "run_id", "AIRFLOW_CONTEXT_DAG_RUN_ID"),
      ]
  
      context_params = settings.get_airflow_context_vars(context)
      for key, value in context_params.items():
          if not isinstance(key, str):
              raise TypeError(f"key <{key}> must be string")
          if not isinstance(value, str):
              raise TypeError(
                  f"value of key <{key}> must be string, not {type(value)}"
              )
  
          if in_env_var_format:
              if not key.startswith(ENV_VAR_FORMAT_PREFIX):
                  key = ENV_VAR_FORMAT_PREFIX + key.upper()
          else:
              if not key.startswith(DEFAULT_FORMAT_PREFIX):
                  key = DEFAULT_FORMAT_PREFIX + key
          params[key] = value
  
      for subject, attr, mapping_key in ops:
          _attr = getattr(subject, attr, None)
          if subject and _attr:
              mapping_value = AIRFLOW_VAR_NAME_FORMAT_MAPPING[
                  mapping_key
              ][name_format]
              if isinstance(_attr, str):
                  params[mapping_value] = _attr
              elif isinstance(_attr, datetime):
                  params[mapping_value] = _attr.isoformat()
              elif isinstance(_attr, list):
                  # os env variable value needs to be string
                  params[mapping_value] = ",".join(_attr)
              else:
                  params[mapping_value] = str(_attr)
  
      return params
  
file_path: airflow/airflow/utils/operator_helpers.py
imports: from __future__ import annotations
  from datetime import datetime
  from typing import Any, Callable, Collection, Mapping, TypeVar
  from airflow import settings
  from airflow.utils.context import Context, lazy_mapping_from_context
label_desc: Use image node_id 1 for calling a global function with example usage: context_to_airflow_vars(context, in_env_var_format) and returns: params