code: def _setup_source_conn(self, source_conn_id, source_bucket_name=None):
          
          self.source_conn = BaseHook.get_hook(source_conn_id)
          self.source_conn_id = source_conn_id
  
          # Workaround for getting hook in case of s3 connection
          # This is needed because get_hook silently returns None for s3 connections
          # See https://issues.apache.org/jira/browse/AIRFLOW-2316 for more info
          connection = BaseHook._get_connection_from_env(source_conn_id)
          self.log.info(connection.extra_dejson)
          if connection.conn_type == 's3':
              self.log.info("Setting up s3 connection {0}".format(source_conn_id))
              self.source_conn = S3Hook(aws_conn_id=source_conn_id)
              # End Workaround
              if source_bucket_name is None:
                  raise AttributeError("Missing source bucket for s3 connection")
              self.source_bucket_name = source_bucket_name
  
          if not isinstance(self.source_conn, DbApiHook) and not isinstance(self.source_conn, S3Hook):
              raise AttributeError(
                  "Only s3_csv, local and sql connection types are allowed, not {0}".format(type(self.source_conn)))
docs: Retrieve connection based on source_conn_id. In case of s3 it also configures the bucket.
          Validates that connection id belongs to supported connection type.
          :param source_conn_id:
          :param source_bucket_name: