Skip to content

delete_event_bus

delete_event_bus(
    name: EventBusName, *, config_overrides: Optional[EventBridgeClientConfig] = None
) -> None

Deletes the specified custom event bus or partner event bus. All rules associated with this event bus need to be deleted. You can't delete your account's default event bus.

Parameters:

  • name (EventBusName) –

    The name of the event bus to delete.

Raises:

Source code in docs_src/capo_eventbridge/_services/event_bridge.py
def delete_event_bus(
    self,
    name: "capo_eventbridge.types.event_bus_name.EventBusName",
    *,
    config_overrides: Optional[EventBridgeClientConfig] = None,
) -> None:
    """<p>Deletes the specified custom event bus or partner event bus. All rules associated with this event bus need to be deleted. You can't delete your account's default event bus.</p>

    Args:
        name: <p>The name of the event bus to delete.</p>

    Raises:
        capo_eventbridge.errors.concurrent_modification_exception.ConcurrentModificationException: <p>There is concurrent modification on a rule, target, archive, or replay.</p>
        capo_eventbridge.errors.internal_exception.InternalException: <p>This exception occurs due to unexpected causes.</p>
        capo_eventbridge.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_eventbridge.types.delete_event_bus_request.DeleteEventBusRequest]",
    ) -> OperationResponse[None]:
        import capo_eventbridge._operations.aws_events.delete_event_bus

        output, http_response = (
            capo_eventbridge._operations.aws_events.delete_event_bus.delete_event_bus(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    interceptors_, options_ = self.operation_options(
        config_overrides, OperationLogging("DeleteEventBus")
    )
    input_: capo_eventbridge.types.delete_event_bus_request.DeleteEventBusRequest = {
        "name": name
    }

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