https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L145
"""
    Set task instance duration.
    :param task_instance: the task instance
    :meta private:
    """
    if task_instance.end_date and task_instance.start_date:
        task_instance.duration = (task_instance.end_date - task_instance.start_date).total_seconds()
    else:
        task_instance.duration = None
    log.debug("Task Duration set to %s", task_instance.duration)
def _stats_tags(*, task_instance: TaskInstance | TaskInstancePydantic) -> dict[str, str]:
    """
    Returns task instance tags.
    :param task_instance: the task instance
    :meta private:
    """
    return prune_dict({"dag_id": task_instance.dag_id, "task_id": task_instance.task_id})
def _clear_next_method_args(*, task_instance: TaskInstance | TaskInstancePydantic) -> None:
    """
    Ensure we unset next_method and next_kwargs to ensure that any retries don't reuse them.
    :param task_instance: the task instance
    :meta private:
    """
    log.debug("Clearing next_method and next_kwargs.")
    task_instance.next_method = None
    task_instance.next_kwargs = None
def _get_template_context(
    *,
    task_instance,
    session: Session | None = None,