class_docstr: Executes a Spark job on Azure Synapse.
  
  .. see also::
      For more information on how to use this operator, take a look at the guide:
      :ref:`howto/operator:AzureSynapseRunSparkBatchOperator`
  
  :param azure_synapse_conn_id: The connection identifier for connecting to Azure Synapse.
  :param wait_for_termination: Flag to wait on a job run's termination.
  :param spark_pool: The target synapse spark pool used to submit the job
  :param payload: Livy compatible payload which represents the spark job that a user wants to submit
  :param timeout: Time in seconds to wait for a job to reach a terminal status for non-asynchronous
      waits. Used only if ``wait_for_termination`` is True.
  :param check_interval: Time in seconds to check on a job run's status for non-asynchronous waits.
      Used only if ``wait_for_termination`` is True.
code: def execute(self, context: Context) -> None:
      self.log.info("Executing the Synapse spark job.")
      response = self.hook.run_spark_job(payload=self.payload)
      self.log.info(response)
      self.job_id = vars(response)["id"]
      # Push the ``job_id`` value to XCom regardless of what happens during execution. This allows for
      # retrieval the executed job's ``id`` for downstream tasks especially if performing an
      # asynchronous wait.
      context["ti"].xcom_push(key="job_id", value=self.job_id)
  
      if self.wait_for_termination:
          self.log.info(
              "Waiting for job run %s to terminate.", self.job_id
          )
  
          if self.hook.wait_for_job_run_status(
              job_id=self.job_id,
              expected_statuses=AzureSynapseSparkBatchRunStatus.SUCCESS,
              check_interval=self.check_interval,
              timeout=self.timeout,
          ):
              self.log.info(
                  "Job run %s has completed successfully.", self.job_id
              )
          else:
              raise Exception(
                  f"Job run {self.job_id} has failed or has been cancelled."
              )
  
imports: from __future__ import annotations
  from functools import cached_property
  from typing import TYPE_CHECKING, Any, Sequence
  from urllib.parse import urlencode
  from airflow.exceptions import AirflowException
  from airflow.hooks.base import BaseHook
  from airflow.models import BaseOperator, BaseOperatorLink, XCom
  from airflow.providers.microsoft.azure.hooks.synapse import AzureSynapseHook, AzureSynapsePipelineHook, AzureSynapsePipelineRunException, AzureSynapsePipelineRunStatus, AzureSynapseSparkBatchRunStatus
label_desc: Use image node_id 3 for calling the AzureSynapseRunSparkBatchOperator obj's underlying member method code with example usage: obj.execute(context) without return types