create_channel ¶

def create_channel(
    self,
    cluster_arn: "capo_kafka.types.__string.__string",
    *,
    config_overrides: Optional[KafkaClientConfig] = None,
    channel_name: Optional["capo_kafka.types.__string.__string"] = None,
    encryption_configuration: Optional[
        "capo_kafka.types.encryption_configuration.EncryptionConfiguration"
    ] = None,
    iceberg_destination_configuration: Optional[
        "capo_kafka.types.iceberg_destination_configuration.IcebergDestinationConfiguration"
    ] = None,
    s3_destination_configuration: Optional[
        "capo_kafka.types.s3_destination_configuration.S3DestinationConfiguration"
    ] = None,
    tags: Optional["capo_kafka.types.__map_of__string.__mapOf__string"] = None,
    topic_configuration_list: Optional[
        "capo_kafka.types.__list_of_topic_configuration.__listOfTopicConfiguration"
    ] = None,
    logging_info: Optional[
        "capo_kafka.types.channel_logging_info.ChannelLoggingInfo"
    ] = None,
) -> "capo_kafka.types.create_channel_response.CreateChannelResponse":
    """<p>Creates a Channel that streams records from an Amazon MSK Express cluster topic to Amazon S3 or Apache Iceberg.</p>

    Args:
        channel_name: <p>The name of the channel. Must be unique within the cluster.</p>
        cluster_arn: <p>The Amazon Resource Name (ARN) that uniquely identifies the cluster.</p>
        encryption_configuration: <p>The encryption configuration applied to the channel.</p>
        iceberg_destination_configuration: <p>The Apache Iceberg destination for the channel. Mutually exclusive with s3DestinationConfiguration.</p>
        s3_destination_configuration: <p>The Amazon S3 destination for the channel. Mutually exclusive with icebergDestinationConfiguration.</p>
        tags: <p>The tags attached to the channel.</p>
        topic_configuration_list: <p>The list of topic configurations for the channel. Currently exactly one topic must be specified.</p>
        logging_info: <p>The destinations to which the channel publishes operational logs.</p>

    Raises:
        capo_kafka.errors.bad_request_exception.BadRequestException: <p>Returns information about an error.</p>
        capo_kafka.errors.conflict_exception.ConflictException: <p>Returns information about an error.</p>
        capo_kafka.errors.forbidden_exception.ForbiddenException: <p>Returns information about an error.</p>
        capo_kafka.errors.internal_server_error_exception.InternalServerErrorException: <p>Returns information about an error.</p>
        capo_kafka.errors.not_found_exception.NotFoundException: <p>Returns information about an error.</p>
        capo_kafka.errors.service_unavailable_exception.ServiceUnavailableException: <p>Returns information about an error.</p>
        capo_kafka.errors.too_many_requests_exception.TooManyRequestsException: <p>Returns information about an error.</p>
        capo_kafka.errors.unauthorized_exception.UnauthorizedException: <p>Returns information about an error.</p>
        capo_kafka.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_kafka.types.create_channel_request.CreateChannelRequest]",
    ) -> OperationResponse[
        "capo_kafka.types.create_channel_response.CreateChannelResponse"
    ]:
        import capo_kafka._operations.kafka.create_channel

        output, http_response = (
            capo_kafka._operations.kafka.create_channel.create_channel(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("CreateChannel")
    )
    input_: capo_kafka.types.create_channel_request.CreateChannelRequest = {
        "cluster_arn": cluster_arn
    }
    if channel_name is not None:
        input_["channel_name"] = channel_name
    if encryption_configuration is not None:
        input_["encryption_configuration"] = encryption_configuration
    if iceberg_destination_configuration is not None:
        input_["iceberg_destination_configuration"] = (
            iceberg_destination_configuration
        )
    if s3_destination_configuration is not None:
        input_["s3_destination_configuration"] = s3_destination_configuration
    if tags is not None:
        input_["tags"] = tags
    if topic_configuration_list is not None:
        input_["topic_configuration_list"] = topic_configuration_list
    if logging_info is not None:
        input_["logging_info"] = logging_info

    response = execute_pipeline(
        OperationRequest(input=input_, options=options_),
        handler=_handler,
        interceptors=list(interceptors_),
    )
    response.response.close()
    return response.output