class_docstr: Google Storage Transfer Service functionalities.
  
  All methods in the hook with *project_id* in the signature must be called
  with keyword arguments rather than positional.
docstr: Wait until the job reaches the expected state.
  
  :param job: The transfer job to wait for. See:
      https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs#TransferJob
  :param expected_statuses: The expected state. See:
      https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferOperations#Status
  :param timeout: Time in which the operation must end in seconds. If not
      specified, defaults to 60 seconds.
code: def wait_for_transfer_job(
      self,
      job: dict,
      expected_statuses: set[str] | None = None,
      timeout: float | timedelta | None = None,
  ) -> None:
      """Wait until the job reaches the expected state.
  
      :param job: The transfer job to wait for. See:
          https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs#TransferJob
      :param expected_statuses: The expected state. See:
          https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferOperations#Status
      :param timeout: Time in which the operation must end in seconds. If not
          specified, defaults to 60 seconds.
      """
      expected_statuses = (
          {GcpTransferOperationStatus.SUCCESS}
          if not expected_statuses
          else expected_statuses
      )
      if timeout is None:
          timeout = 60
      elif isinstance(timeout, timedelta):
          timeout = timeout.total_seconds()
  
      start_time = time.monotonic()
      while time.monotonic() - start_time < timeout:
          request_filter = {
              FILTER_PROJECT_ID: job[PROJECT_ID],
              FILTER_JOB_NAMES: [job[NAME]],
          }
          operations = self.list_transfer_operations(
              request_filter=request_filter
          )
  
          for operation in operations:
              self.log.info(
                  "Progress for operation %s: %s",
                  operation[NAME],
                  operation[METADATA][COUNTERS],
              )
  
          if self.operations_contain_expected_statuses(
              operations, expected_statuses
          ):
              return
          time.sleep(TIME_TO_SLEEP_IN_SECONDS)
      raise AirflowException(
          "Timeout. The operation could not be completed within the allotted time."
      )
  
imports: from __future__ import annotations
  import json
  import logging
  import time
  import warnings
  from copy import deepcopy
  from datetime import timedelta
  from typing import TYPE_CHECKING, Any, Sequence
  from google.cloud.storage_transfer_v1 import ListTransferJobsRequest, StorageTransferServiceAsyncClient, TransferJob, TransferOperation
  from googleapiclient.discovery import Resource, build
  from googleapiclient.errors import HttpError
  from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
  from airflow.providers.google.common.hooks.base_google import GoogleBaseAsyncHook, GoogleBaseHook
label_desc: Use image node_id 14 for calling the CloudDataTransferServiceHook obj's underlying member method code with example usage: obj.wait_for_transfer_job(job, expected_statuses, timeout) without return types