def _send_dead_letter_index()

in shippers/es.py [0:0]

• 28 lines of code
• 11 McCabe index (conditional complexity)

    def _send_dead_letter_index(self, actions: list[Any]) -> list[Any]:
        """
        Index the failed actions in the dead letter index (DLI).

        This function attempts to index failed actions to the DLI, but may not do so
        for one of the following reasons:

        1. The failed action could not be encoded for indexing in the DLI.
        2. ES returned an error on the attempt to index the failed action in the DLI.
        3. The failed action error is retryable (connection error or status code 429).

        Retryable errors are not indexed in the DLI, as they are expected to be
        sent again to the data stream at `es_datastream_name` by the replay handler.

        Args:
            actions (list[Any]): A list of actions to index in the DLI.

        Returns:
            list[Any]: A list of actions that were not indexed in the DLI due to one of
            the reasons mentioned above.
        """
        non_indexed_actions: list[Any] = []
        encoded_actions = []

        for action in actions:
            if (
                "http" not in action  # no http status: connection error
                or action["http"]["response"]["status_code"] in _retryable_http_status_codes
            ):
                # We don't want to forward this action to
                # the dead letter index.
                #
                # Add the action to the list of non-indexed
                # actions and continue with the next one.
                non_indexed_actions.append(action)
                continue

            # Reshape event to dead letter index
            encoded = self._encode_dead_letter(action)
            if not encoded:
                shared_logger.error("cannot encode dead letter index event from payload", extra={"action": action})
                non_indexed_actions.append(action)

            encoded_actions.append(encoded)

        # If no action can be encoded, return original action list as failed
        if len(encoded_actions) == 0:
            return non_indexed_actions

        errors = es_bulk(self._es_client, encoded_actions, **self._bulk_kwargs)
        failed = self._handle_outcome(actions=encoded_actions, errors=errors)

        if not isinstance(failed, list) or len(failed) == 0:
            return non_indexed_actions

        for action in failed:
            event_payload = self._decode_dead_letter(action)

            if not event_payload:
                shared_logger.error("cannot decode dead letter index event from payload", extra={"action": action})
                continue

            non_indexed_actions.append(event_payload)

        return non_indexed_actions