def checkpoint()

in amazon_kclpy/kcl.py [0:0]

• 9 lines of code
• 4 McCabe index (conditional complexity)

    def checkpoint(self, sequence_number=None, sub_sequence_number=None):
        """
        Checkpoints at a particular sequence number you provide or if no sequence number is given, the checkpoint will
        be at the end of the most recently delivered list of records

        :param str or None sequence_number: The sequence number to checkpoint at or None if you want to checkpoint at the
            farthest record
        :param int or None sub_sequence_number: the sub sequence to checkpoint at, if set to None will checkpoint
            at the farthest sub_sequence_number
        """
        response = {"action": "checkpoint", "sequenceNumber": sequence_number, "subSequenceNumber": sub_sequence_number}
        self.io_handler.write_action(response)
        action = self._get_action()
        if isinstance(action, messages.CheckpointInput):
            if action.error is not None:
                raise CheckpointError(action.error)
        else:
            #
            # We are in an invalid state. We will raise a checkpoint exception
            # to the RecordProcessor indicating that the KCL (or KCLpy) is in
            # an invalid state. See KCL documentation for description of this
            # exception. Note that the documented guidance is that this exception
            # is NOT retryable so the client code should exit.
            #
            raise CheckpointError('InvalidStateException')