get_execution_history ¶

def get_execution_history(
    self,
    execution_arn: "capo_sfn.types.arn.Arn",
    *,
    config_overrides: Optional[SFNClientConfig] = None,
    max_results: Optional["capo_sfn.types.page_size.PageSize"] = None,
    reverse_order: Optional["capo_sfn.types.reverse_order.ReverseOrder"] = None,
    next_token: Optional["capo_sfn.types.page_token.PageToken"] = None,
    include_execution_data: Optional[
        "capo_sfn.types.include_execution_data_get_execution_history.IncludeExecutionDataGetExecutionHistory"
    ] = None,
) -> "capo_sfn.types.get_execution_history_output.GetExecutionHistoryOutput":
    """<p>Returns the history of the specified execution as a list of events. By default, the results are returned in ascending order of the <code>timeStamp</code> of the events. Use the <code>reverseOrder</code> parameter to get the latest events first.</p> <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 <i>HTTP 400 InvalidToken</i> error.</p> <p>This API action is not supported by <code>EXPRESS</code> state machines.</p>

    Args:
        execution_arn: <p>The Amazon Resource Name (ARN) of the execution.</p>
        max_results: <p>The maximum number of results that are returned per call. You can use <code>nextToken</code> to obtain further pages of results. The default is 100 and the maximum allowed page size is 1000. A value of 0 uses the default.</p> <p>This is only an upper limit. The actual number of results returned per call might be fewer than the specified maximum.</p>
        reverse_order: <p>Lists events in descending order of their <code>timeStamp</code>.</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 <i>HTTP 400 InvalidToken</i> error.</p>
        include_execution_data: <p>You can select whether execution data (input or output of a history event) is returned. The default is <code>true</code>.</p>

    Raises:
        capo_sfn.errors.execution_does_not_exist.ExecutionDoesNotExist: <p>The specified execution does not exist.</p>
        capo_sfn.errors.invalid_arn.InvalidArn: <p>The provided Amazon Resource Name (ARN) is not valid.</p>
        capo_sfn.errors.invalid_token.InvalidToken: <p>The provided token is not valid.</p>
        capo_sfn.errors.kms_access_denied_exception.KmsAccessDeniedException: <p>Either your KMS key policy or API caller does not have the required permissions.</p>
        capo_sfn.errors.kms_invalid_state_exception.KmsInvalidStateException: <p>The KMS key is not in valid state, for example: Disabled or Deleted.</p>
        capo_sfn.errors.kms_throttling_exception.KmsThrottlingException: <p>Received when KMS returns <code>ThrottlingException</code> for a KMS call that Step Functions makes on behalf of the caller.</p>
        capo_sfn.errors.UnknownServiceError: The service returned an error code this client does not model.
    """

    def _handler(
        req: "OperationRequest[capo_sfn.types.get_execution_history_input.GetExecutionHistoryInput]",
    ) -> OperationResponse[
        "capo_sfn.types.get_execution_history_output.GetExecutionHistoryOutput"
    ]:
        import capo_sfn._operations.aws_step_functions.get_execution_history

        output, http_response = (
            capo_sfn._operations.aws_step_functions.get_execution_history.get_execution_history(
                req.options, req.input
            )
        )
        return OperationResponse(output=output, response=http_response)

    def _redact_output(
        output: "capo_sfn.types.get_execution_history_output.GetExecutionHistoryOutput",
    ) -> dict[str, Any]:
        import capo_sfn.types.get_execution_history_output

        return capo_sfn.types.get_execution_history_output.redact_sensitive(output)

    interceptors_, options_ = self.operation_options(
        config_overrides,
        OperationLogging("GetExecutionHistory", redact_output=_redact_output),
    )
    input_: capo_sfn.types.get_execution_history_input.GetExecutionHistoryInput = {
        "execution_arn": execution_arn
    }
    if max_results is not None:
        input_["max_results"] = max_results
    if reverse_order is not None:
        input_["reverse_order"] = reverse_order
    if next_token is not None:
        input_["next_token"] = next_token
    if include_execution_data is not None:
        input_["include_execution_data"] = include_execution_data

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