Method _serialize_field_value

st2common/st2common/fields.py:457–491

Serialize and encode the provided field value.

(self, value: dict)

        return data

    def _serialize_field_value(self, value: dict) -> bytes:
        """
        Serialize and encode the provided field value.
        """
        # Orquesta workflows support toSet() YAQL operator which returns a set which used to get
        # serialized to list by mongoengine DictField.
        #
        # For backward compatibility reasons, we need to support serializing set to a list as
        # well.
        #
        # Based on micro benchmarks, using default function adds very little overhead (1%) so it
        # should be safe to use default for every operation.
        #
        # If this turns out to be not true or it adds more overhead in other scenarios, we should
        # revisit this decision and only use "default" argument where needed (aka Workflow models).
        def default(obj):
            if isinstance(obj, set):
                return list(obj)
            raise TypeError

        if not self.use_header:
            return orjson.dumps(value, default=default)

        data = orjson.dumps(value, default=default)

        if self.compression_algorithm == "zstandard":
            # NOTE: At this point zstandard is only test dependency
            import zstandard

            compression_header = JSONDictFieldCompressionAlgorithmEnum.ZSTANDARD
            data = zstandard.ZstdCompressor().compress(data)
        else:
            compression_header = JSONDictFieldCompressionAlgorithmEnum.NONE

        return compression_header.value + b":" + b"o:" + data

    def __get__(self, instance, owner):
        """