class_docstr: Execute an UNLOAD command to s3 as a CSV with headers.
  
  .. seealso::
      For more information on how to use this operator, take a look at the guide:
      :ref:`howto/operator:RedshiftToS3Operator`
  
  :param s3_bucket: reference to a specific S3 bucket
  :param s3_key: reference to a specific S3 key. If ``table_as_file_name`` is set
      to False, this param must include the desired file name
  :param schema: reference to a specific schema in redshift database
      Applicable when ``table`` param provided.
  :param table: reference to a specific table in redshift database
      Used when ``select_query`` param not provided.
  :param select_query: custom select query to fetch data from redshift database
  :param redshift_conn_id: reference to a specific redshift database
  :param aws_conn_id: reference to a specific S3 connection
      If the AWS connection contains 'aws_iam_role' in ``extras``
      the operator will use AWS STS credentials with a token
      https://docs.aws.amazon.com/redshift/latest/dg/copy-parameters-authorization.html#copy-credentials
  :param verify: Whether or not to verify SSL certificates for S3 connection.
      By default SSL certificates are verified.
      You can provide the following values:
  
      - ``False``: do not validate SSL certificates. SSL will still be used
               (unless use_ssl is False), but SSL certificates will not be
               verified.
      - ``path/to/cert/bundle.pem``: A filename of the CA cert bundle to uses.
               You can specify this argument if you want to use a different
               CA cert bundle than the one used by botocore.
  :param unload_options: reference to a list of UNLOAD options
  :param autocommit: If set to True it will automatically commit the UNLOAD statement.
      Otherwise it will be committed right before the redshift connection gets closed.
  :param include_header: If set to True the s3 file contains the header columns.
  :param parameters: (optional) the parameters to render the SQL query with.
  :param table_as_file_name: If set to True, the s3 file will be named as the table.
      Applicable when ``table`` param provided.
  :param redshift_data_api_kwargs: If using the Redshift Data API instead of the SQL-based connection,
      dict of arguments for the hook's ``execute_query`` method.
      Cannot include any of these kwargs: ``{'sql', 'parameters'}``
code: def __init__(
      self,
      *,
      s3_bucket: str,
      s3_key: str,
      schema: str | None = None,
      table: str | None = None,
      select_query: str | None = None,
      redshift_conn_id: str = "redshift_default",
      aws_conn_id: str = "aws_default",
      verify: bool | str | None = None,
      unload_options: list | None = None,
      autocommit: bool = False,
      include_header: bool = False,
      parameters: Iterable | Mapping | None = None,
      table_as_file_name: bool = True,  # Set to True by default for not breaking current workflows
      redshift_data_api_kwargs: dict | None = None,
      **kwargs,
  ) -> None:
      super().__init__(**kwargs)
      self.s3_bucket = s3_bucket
      self.s3_key = (
          f"{s3_key}/{table}_"
          if (table and table_as_file_name)
          else s3_key
      )
      self.schema = schema
      self.table = table
      self.redshift_conn_id = redshift_conn_id
      self.aws_conn_id = aws_conn_id
      self.verify = verify
      self.unload_options: list = unload_options or []
      self.autocommit = autocommit
      self.include_header = include_header
      self.parameters = parameters
      self.table_as_file_name = table_as_file_name
      self.redshift_data_api_kwargs = redshift_data_api_kwargs or {}
  
      if select_query:
          self.select_query = select_query
      elif self.schema and self.table:
          self.select_query = (
              f"SELECT * FROM {self.schema}.{self.table}"
          )
      else:
          raise ValueError(
              "Please provide both `schema` and `table` params or `select_query` to fetch the data."
          )
  
      if self.include_header and "HEADER" not in [
          uo.upper().strip() for uo in self.unload_options
      ]:
          self.unload_options = [*self.unload_options, "HEADER"]
  
      if self.redshift_data_api_kwargs:
          for arg in ["sql", "parameters"]:
              if arg in self.redshift_data_api_kwargs:
                  raise AirflowException(
                      f"Cannot include param '{arg}' in Redshift Data API kwargs"
                  )
  
file_path: airflow/airflow/providers/amazon/aws/transfers/redshift_to_s3.py
imports: from __future__ import annotations
  import re
  from typing import TYPE_CHECKING, Iterable, Mapping, Sequence
  from airflow.exceptions import AirflowException
  from airflow.models import BaseOperator
  from airflow.providers.amazon.aws.hooks.redshift_data import RedshiftDataHook
  from airflow.providers.amazon.aws.hooks.redshift_sql import RedshiftSQLHook
  from airflow.providers.amazon.aws.hooks.s3 import S3Hook
  from airflow.providers.amazon.aws.utils.redshift import build_credentials_block
label_desc: Use image node_id 1 to create a new RedshiftToS3Operator object from inherited base classes: BaseOperator with example: obj = RedshiftToS3Operator()