def serialize()

in python/src/pywy/core/serializer.py [0:0]

• 41 lines of code
• 16 McCabe index (conditional complexity)

    def serialize(self, operator):
        json_operator = {}
        json_operator["id"] = self.id_table[operator]
        json_operator["operatorName"] = operator.json_name
        json_operator["cat"] = operator.cat
        if operator.cat != "input":
            json_operator["input"] = list(map(lambda x: self.id_table[x], operator.inputOperator))
        else:
            json_operator["input"] = []

        if operator.cat != "output":
            json_operator["output"] = list(map(lambda x: self.id_table[x], operator.outputOperator))
        else:
            json_operator["output"] = []

        json_operator["data"] = {}

        if hasattr(operator, "input_type") and operator.input_type is not None:
            json_operator["data"]["inputType"] = ndim_from_type(operator.input_type).to_json()

        if hasattr(operator, "output_type") and operator.output_type is not None:
            json_operator["data"]["outputType"] = ndim_from_type(operator.output_type).to_json()

        if operator.json_name == "filter":
            json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.use_predicate)).decode('utf-8')

        elif operator.json_name == "reduceBy":
            json_operator["data"]["keyUdf"] = base64.b64encode(cloudpickle.dumps(operator.key_function)).decode('utf-8')
            json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.reduce_function)).decode('utf-8')

        elif operator.json_name == "join":
            json_operator["data"]["thisKeyUdf"] = base64.b64encode(cloudpickle.dumps(operator.this_key_function)).decode('utf-8')
            json_operator["data"]["thatKeyUdf"] = base64.b64encode(cloudpickle.dumps(operator.that_key_function)).decode('utf-8')

        elif operator.json_name == "cartesian":
            del json_operator["data"]

        elif operator.json_name == "dlTraining":
            json_operator["data"]["model"] = {"modelType": "DLModel", "op": operator.model.get_out().to_dict()}
            json_operator["data"]["option"] = operator.option.to_dict()

        else:
            if hasattr(operator, "get_udf"):
                json_operator["data"]["udf"] = base64.b64encode(cloudpickle.dumps(operator.get_udf)).decode('utf-8')

            if hasattr(operator, "path"):
                json_operator["data"]["filename"] = operator.path

            if hasattr(operator, "projection"):
                json_operator["data"]["projection"] = operator.projection

            if hasattr(operator, "column_names"):
                json_operator["data"]["column_names"] = operator.column_names

        return json_operator