put_events ¶

def put_events(
    self,
    entries: "capo_cloudwatch_events.types.put_events_request_entry_list.PutEventsRequestEntryList",
    *,
    config_overrides: Optional[CloudWatchEventsClientConfig] = None,
) -> "capo_cloudwatch_events.types.put_events_response.PutEventsResponse":
    """<p>Sends custom events to Amazon EventBridge so that they can be matched to rules.</p>

    Args:
        entries: <p>The entry that defines an event in your system. You can specify several parameters for the entry such as the source and type of the event, resources associated with the event, and so on.</p>

    Raises:
        capo_cloudwatch_events.errors.internal_exception.InternalException: <p>This exception occurs due to unexpected causes.</p>
        capo_cloudwatch_events.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_cloudwatch_events.types.put_events_request.PutEventsRequest]",
    ) -> OperationResponse[
        "capo_cloudwatch_events.types.put_events_response.PutEventsResponse"
    ]:
        import capo_cloudwatch_events._operations.aws_events.put_events

        output, http_response = (
            capo_cloudwatch_events._operations.aws_events.put_events.put_events(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("PutEvents")
    )
    input_: capo_cloudwatch_events.types.put_events_request.PutEventsRequest = {
        "entries": entries
    }

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