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: Create a transfer job that runs periodically.
  
  :param body: (Required) The request body, as described in
      https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs/patch#request-body
  :return: The transfer job. See:
      https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs#TransferJob
code: def create_transfer_job(self, body: dict) -> dict:
      """Create a transfer job that runs periodically.
  
      :param body: (Required) The request body, as described in
          https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs/patch#request-body
      :return: The transfer job. See:
          https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferJobs#TransferJob
      """
      body = self._inject_project_id(body, BODY, PROJECT_ID)
  
      try:
          transfer_job = (
              self.get_conn()
              .transferJobs()
              .create(body=body)
              .execute(num_retries=self.num_retries)
          )
      except HttpError as e:
          # If status code "Conflict"
          # https://cloud.google.com/storage-transfer/docs/reference/rest/v1/transferOperations#Code.ENUM_VALUES.ALREADY_EXISTS
          # we should try to find this job
          job_name = body.get(JOB_NAME, "")
          if int(e.resp.status) == ALREADY_EXIST_CODE and job_name:
              transfer_job = self.get_transfer_job(
                  job_name=job_name, project_id=body.get(PROJECT_ID)
              )
              # Generate new job_name, if jobs status is deleted
              # and try to create this job again
              if (
                  transfer_job.get(STATUS)
                  == GcpTransferJobsStatus.DELETED
              ):
                  body[JOB_NAME] = gen_job_name(job_name)
                  self.log.info(
                      "Job `%s` has been soft deleted. Creating job with new name `%s`",
                      job_name,
                      {body[JOB_NAME]},
                  )
  
                  return (
                      self.get_conn()
                      .transferJobs()
                      .create(body=body)
                      .execute(num_retries=self.num_retries)
                  )
              elif (
                  transfer_job.get(STATUS)
                  == GcpTransferJobsStatus.DISABLED
              ):
                  return self.enable_transfer_job(
                      job_name=job_name, project_id=body.get(PROJECT_ID)
                  )
          else:
              raise e
      self.log.info("Created job %s", transfer_job[NAME])
      return transfer_job
  
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 3 for calling the CloudDataTransferServiceHook obj's underlying member method code with example usage: obj.create_transfer_job(body) and returns: transfer_job, self, self