list_pipes ¶

def list_pipes(
    self,
    *,
    config_overrides: Optional[PipesClientConfig] = None,
    name_prefix: Optional["capo_pipes.types.pipe_name.PipeName"] = None,
    desired_state: Optional[
        "capo_pipes.types.requested_pipe_state.RequestedPipeState"
    ] = None,
    current_state: Optional["capo_pipes.types.pipe_state.PipeState"] = None,
    source_prefix: Optional["capo_pipes.types.resource_arn.ResourceArn"] = None,
    target_prefix: Optional["capo_pipes.types.resource_arn.ResourceArn"] = None,
    next_token: Optional["capo_pipes.types.next_token.NextToken"] = None,
    limit: Optional["capo_pipes.types.limit_max100.LimitMax100"] = None,
) -> "capo_pipes.types.list_pipes_response.ListPipesResponse":
    """<p>Get the pipes associated with this account. For more information about pipes, see <a href="https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-pipes.html">Amazon EventBridge Pipes</a> in the Amazon EventBridge User Guide.</p>

    Args:
        name_prefix: <p>A value that will return a subset of the pipes associated with this account. For example, <code>"NamePrefix": "ABC"</code> will return all endpoints with "ABC" in the name.</p>
        desired_state: <p>The state the pipe should be in.</p>
        current_state: <p>The state the pipe is in.</p>
        source_prefix: <p>The prefix matching the pipe source.</p>
        target_prefix: <p>The prefix matching the pipe target.</p>
        next_token: <p>If <code>nextToken</code> is returned, there are more results available. The value of <code>nextToken</code> is a unique pagination token for each page. Make the call again using the returned token to retrieve the next page. Keep all other arguments unchanged. Each pagination token expires after 24 hours. Using an expired pagination token will return an HTTP 400 InvalidToken error.</p>
        limit: <p>The maximum number of pipes to include in the response.</p>

    Raises:
        capo_pipes.errors.internal_exception.InternalException: <p>This exception occurs due to unexpected causes.</p>
        capo_pipes.errors.throttling_exception.ThrottlingException: <p>An action was throttled.</p>
        capo_pipes.errors.validation_exception.ValidationException: <p>Indicates that an error has occurred while performing a validate operation.</p>
        capo_pipes.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_pipes.types.list_pipes_request.ListPipesRequest]",
    ) -> OperationResponse[
        "capo_pipes.types.list_pipes_response.ListPipesResponse"
    ]:
        import capo_pipes._operations.pipes.list_pipes

        output, http_response = capo_pipes._operations.pipes.list_pipes.list_pipes(
            req.options, req.input
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_input(
        input_: "capo_pipes.types.list_pipes_request.ListPipesRequest",
    ) -> dict[str, Any]:
        import capo_pipes.types.list_pipes_request

        return capo_pipes.types.list_pipes_request.redact_sensitive(input_)

    def _redact_output(
        output: "capo_pipes.types.list_pipes_response.ListPipesResponse",
    ) -> dict[str, Any]:
        import capo_pipes.types.list_pipes_response

        return capo_pipes.types.list_pipes_response.redact_sensitive(output)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging(
            "ListPipes", redact_input=_redact_input, redact_output=_redact_output
        ),
    )
    input_: capo_pipes.types.list_pipes_request.ListPipesRequest = {}
    if name_prefix is not None:
        input_["name_prefix"] = name_prefix
    if desired_state is not None:
        input_["desired_state"] = desired_state
    if current_state is not None:
        input_["current_state"] = current_state
    if source_prefix is not None:
        input_["source_prefix"] = source_prefix
    if target_prefix is not None:
        input_["target_prefix"] = target_prefix
    if next_token is not None:
        input_["next_token"] = next_token
    if limit is not None:
        input_["limit"] = limit

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