code: def find_task_relatives(tasks, downstream, upstream):
      """Yield task ids and optionally ancestor and descendant ids."""
      for item in tasks:
          if isinstance(item, tuple):
              task, map_index = item
              yield task.task_id, map_index
          else:
              task = item
              yield task.task_id
          if downstream:
              for relative in task.get_flat_relatives(upstream=False):
                  yield relative.task_id
          if upstream:
              for relative in task.get_flat_relatives(upstream=True):
                  yield relative.task_id
  
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Collection, Iterable, Iterator, NamedTuple
  from sqlalchemy import or_, select
  from sqlalchemy.orm import lazyload
  from airflow.models.dagrun import DagRun
  from airflow.models.taskinstance import TaskInstance
  from airflow.operators.subdag import SubDagOperator
  from airflow.utils import timezone
  from airflow.utils.helpers import exactly_one
  from airflow.utils.session import NEW_SESSION, provide_session
  from airflow.utils.state import DagRunState, State, TaskInstanceState
  from airflow.utils.types import DagRunType
label_desc: Use image node_id 8 for calling a global function with example usage: find_task_relatives(tasks, downstream, upstream) without return types