class_docstr: Creates a manual snapshot of the specified cluster. The cluster must be in the available state.
  
  .. seealso::
      For more information on how to use this operator, take a look at the guide:
      :ref:`howto/operator:RedshiftCreateClusterSnapshotOperator`
  
  :param snapshot_identifier: A unique identifier for the snapshot that you are requesting
  :param cluster_identifier: The cluster identifier for which you want a snapshot
  :param retention_period: The number of days that a manual snapshot is retained.
      If the value is -1, the manual snapshot is retained indefinitely.
  :param tags: A list of tag instances
  :param wait_for_completion: Whether wait for the cluster snapshot to be in ``available`` state
  :param poll_interval: Time (in seconds) to wait between two consecutive calls to check state
  :param max_attempt: The maximum number of attempts to be made to check the state
  :param aws_conn_id: The Airflow connection used for AWS credentials.
      The default connection id is ``aws_default``
  :param deferrable: If True, the operator will run as a deferrable operator.
code: def execute(self, context: Context) -> Any:
      cluster_state = self.redshift_hook.cluster_status(
          cluster_identifier=self.cluster_identifier
      )
      if cluster_state != "available":
          raise AirflowException(
              "Redshift cluster must be in available state. "
              f"Redshift cluster current state is {cluster_state}"
          )
  
      self.redshift_hook.create_cluster_snapshot(
          cluster_identifier=self.cluster_identifier,
          snapshot_identifier=self.snapshot_identifier,
          retention_period=self.retention_period,
          tags=self.tags,
      )
      if self.deferrable:
          self.defer(
              trigger=RedshiftCreateClusterSnapshotTrigger(
                  cluster_identifier=self.cluster_identifier,
                  waiter_delay=self.poll_interval,
                  waiter_max_attempts=self.max_attempt,
                  aws_conn_id=self.aws_conn_id,
              ),
              method_name="execute_complete",
              # timeout is set to ensure that if a trigger dies, the timeout does not restart
              # 60 seconds is added to allow the trigger to exit gracefully (i.e. yield TriggerEvent)
              timeout=timedelta(
                  seconds=self.max_attempt * self.poll_interval + 60
              ),
          )
  
      if self.wait_for_completion:
          self.redshift_hook.get_conn().get_waiter(
              "snapshot_available"
          ).wait(
              ClusterIdentifier=self.cluster_identifier,
              WaiterConfig={
                  "Delay": self.poll_interval,
                  "MaxAttempts": self.max_attempt,
              },
          )
  
imports: from __future__ import annotations
  import time
  from datetime import timedelta
  from typing import TYPE_CHECKING, Any, Sequence
  from airflow.configuration import conf
  from airflow.exceptions import AirflowException
  from airflow.models import BaseOperator
  from airflow.providers.amazon.aws.hooks.redshift_cluster import RedshiftHook
  from airflow.providers.amazon.aws.triggers.redshift_cluster import RedshiftCreateClusterSnapshotTrigger, RedshiftCreateClusterTrigger, RedshiftDeleteClusterTrigger, RedshiftPauseClusterTrigger, RedshiftResumeClusterTrigger
label_desc: Use image node_id 2 for calling the RedshiftCreateClusterSnapshotOperator obj's underlying member method code with example usage: obj.execute(context) without return types