from queue import Queue
from typing import List, Set

from airflow.models.taskinstance import TaskInstance


def get_ancestor_task_ids(ti: TaskInstance) -> List[str]:
    ancestor_task_ids: Set[str] = set()
    tasks_queue: Queue = Queue()
    # determine parent tasks to begin BFS
    for task in ti.task.upstream_list:
        tasks_queue.put(task)
    # perform BFS
    while not tasks_queue.empty():
        task = tasks_queue.get()
        ancestor_task_ids.add(task.task_id)
        for _task in task.upstream_list:
            tasks_queue.put(_task)
    return ancestor_task_ids