def _encode_dead_letter()

in shippers/es.py [0:0]

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

    def _encode_dead_letter(self, outcome: dict[str, Any]) -> dict[str, Any]:
        if "action" not in outcome or "error" not in outcome:
            return {}

        # Assign random id in case bulk() results in error, it can be matched to the original
        # action
        encoded = {
            "@timestamp": datetime.datetime.now(datetime.UTC).strftime("%Y-%m-%dT%H:%M:%S.%fZ"),
            "_id": str(uuid.uuid4()),
            "_index": self._es_dead_letter_index,
            "_op_type": "create",
            "message": json_dumper(outcome["action"]),
            "error": outcome["error"],
        }

        if "http" in outcome:
            # the `http.response.status_code` is not
            # always present in the error field.
            encoded["http"] = outcome["http"]

        return encoded