gremlin-python/src/main/python/gremlin_python/structure/io/graphsonV2d0.py [235:416]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class TraversalStrategySerializer(_GraphSONTypeIO):
    python_type = TraversalStrategy

    @classmethod
    def dictify(cls, strategy, writer):
        strat_dict = {}
        strat_dict["fqcn"] = strategy.fqcn
        strat_dict["conf"] = {}
        for key in strategy.configuration:
            strat_dict["conf"][key] = writer.to_dict(strategy.configuration[key])
        return GraphSONUtil.typed_value(strategy.strategy_name, strat_dict)


class TraverserIO(_GraphSONTypeIO):
    python_type = Traverser
    graphson_type = "g:Traverser"

    @classmethod
    def dictify(cls, traverser, writer):
        return GraphSONUtil.typed_value("Traverser", {"value": writer.to_dict(traverser.object),
                                                     "bulk": writer.to_dict(traverser.bulk)})

    @classmethod
    def objectify(cls, d, reader):
        return Traverser(reader.to_object(d["value"]),
                         reader.to_object(d["bulk"]))


class EnumSerializer(_GraphSONTypeIO):
    python_type = Enum

    @classmethod
    def dictify(cls, enum, _):
        return GraphSONUtil.typed_value(SymbolUtil.to_camel_case(type(enum).__name__),
                                        SymbolUtil.to_camel_case(str(enum.name)))


class PSerializer(_GraphSONTypeIO):
    python_type = P

    @classmethod
    def dictify(cls, p, writer):
        out = {"predicate": p.operator,
               "value": [writer.to_dict(p.value), writer.to_dict(p.other)] if p.other is not None else
               writer.to_dict(p.value)}
        return GraphSONUtil.typed_value("P", out)


class TextPSerializer(_GraphSONTypeIO):
    python_type = TextP

    @classmethod
    def dictify(cls, p, writer):
        out = {"predicate": p.operator,
               "value": [writer.to_dict(p.value), writer.to_dict(p.other)] if p.other is not None else
               writer.to_dict(p.value)}
        return GraphSONUtil.typed_value("TextP", out)


class BindingSerializer(_GraphSONTypeIO):
    python_type = Binding

    @classmethod
    def dictify(cls, binding, writer):
        out = {"key": binding.key,
               "value": writer.to_dict(binding.value)}
        return GraphSONUtil.typed_value("Binding", out)


class LambdaSerializer(_GraphSONTypeIO):
    python_type = FunctionType

    @classmethod
    def dictify(cls, lambda_object, writer):
        lambda_result = lambda_object()
        script = lambda_result if isinstance(lambda_result, str) else lambda_result[0]
        language = statics.default_lambda_language if isinstance(lambda_result, str) else lambda_result[1]
        out = {"script": script,
               "language": language}
        if language == "gremlin-groovy" and "->" in script:
            # if the user has explicitly added parameters to the groovy closure then we can easily detect one or two
            # arg lambdas - if we can't detect 1 or 2 then we just go with "unknown"
            args = script[0:script.find("->")]
            out["arguments"] = 2 if "," in args else 1
        else:
            out["arguments"] = -1

        return GraphSONUtil.typed_value("Lambda", out)


class TypeSerializer(_GraphSONTypeIO):
    python_type = TypeType

    @classmethod
    def dictify(cls, typ, writer):
        return writer.to_dict(typ())


class UUIDIO(_GraphSONTypeIO):
    python_type = uuid.UUID
    graphson_type = "g:UUID"
    graphson_base_type = "UUID"

    @classmethod
    def dictify(cls, obj, writer):
        return GraphSONUtil.typed_value(cls.graphson_base_type, str(obj))

    @classmethod
    def objectify(cls, d, reader):
        return cls.python_type(d)


class DateIO(_GraphSONTypeIO):
    graphson_type = "g:Date"
    graphson_base_type = "Date"

    @classmethod
    def dictify(cls, obj, writer):
        try:
            timestamp_seconds = calendar.timegm(obj.utctimetuple())
            pts = timestamp_seconds * 1e3 + getattr(obj, 'microsecond', 0) / 1e3
        except AttributeError:
            pts = calendar.timegm(obj.timetuple()) * 1e3

        ts = int(round(pts))
        return GraphSONUtil.typed_value(cls.graphson_base_type, ts)

    @classmethod
    def objectify(cls, ts, reader):
        # Python timestamp expects seconds
        return datetime.datetime.utcfromtimestamp(ts / 1000.0)


    class OffsetDateTimeIO(_GraphSONTypeIO):
        python_type = datetime.datetime
        graphson_type = "gx:OffsetDateTime"
        graphson_base_type = "OffsetDateTime"

        @classmethod
        def dictify(cls, obj, writer):
            if obj.tzinfo is None:
                return DateIO.dictify(obj, writer)
            return GraphSONUtil.typed_value(cls.graphson_base_type, obj.isoformat(), "gx")

        @classmethod
        def objectify(cls, dt, reader):
            # specially handling as python isoformat does not support zulu until 3.11
            dt_iso = dt[:-1] + '+00:00' if dt.endswith('Z') else dt
            return datetime.datetime.fromisoformat(dt_iso)


# Based on current implementation, this class must always be declared before FloatIO.
# Seems pretty fragile for future maintainers. Maybe look into this.
class TimestampIO(_GraphSONTypeIO):
    """A timestamp in Python is type float"""
    python_type = statics.timestamp
    graphson_type = "g:Timestamp"
    graphson_base_type = "Timestamp"

    @classmethod
    def dictify(cls, obj, writer):
        # Java timestamp expects milliseconds integer
        # Have to use int because of legacy Python
        ts = int(round(obj * 1000))
        return GraphSONUtil.typed_value(cls.graphson_base_type, ts)

    @classmethod
    def objectify(cls, ts, reader):
        # Python timestamp expects seconds
        return cls.python_type(ts / 1000.0)


class _NumberIO(_GraphSONTypeIO):
    @classmethod
    def dictify(cls, n, writer):
        if isinstance(n, bool):  # because isinstance(False, int) and isinstance(True, int)
            return n
        return GraphSONUtil.typed_value(cls.graphson_base_type, n)

    @classmethod
    def objectify(cls, v, _):
        return cls.python_type(v)
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



gremlin-python/src/main/python/gremlin_python/structure/io/graphsonV3d0.py [239:420]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
class TraversalStrategySerializer(_GraphSONTypeIO):
    python_type = TraversalStrategy

    @classmethod
    def dictify(cls, strategy, writer):
        strat_dict = {}
        strat_dict["fqcn"] = strategy.fqcn
        strat_dict["conf"] = {}
        for key in strategy.configuration:
            strat_dict["conf"][key] = writer.to_dict(strategy.configuration[key])
        return GraphSONUtil.typed_value(strategy.strategy_name, strat_dict)


class TraverserIO(_GraphSONTypeIO):
    python_type = Traverser
    graphson_type = "g:Traverser"

    @classmethod
    def dictify(cls, traverser, writer):
        return GraphSONUtil.typed_value("Traverser", {"value": writer.to_dict(traverser.object),
                                                     "bulk": writer.to_dict(traverser.bulk)})

    @classmethod
    def objectify(cls, d, reader):
        return Traverser(reader.to_object(d["value"]),
                         reader.to_object(d["bulk"]))


class EnumSerializer(_GraphSONTypeIO):
    python_type = Enum

    @classmethod
    def dictify(cls, enum, _):
        return GraphSONUtil.typed_value(SymbolUtil.to_camel_case(type(enum).__name__),
                                        SymbolUtil.to_camel_case(str(enum.name)))


class PSerializer(_GraphSONTypeIO):
    python_type = P

    @classmethod
    def dictify(cls, p, writer):
        out = {"predicate": p.operator,
               "value": [writer.to_dict(p.value), writer.to_dict(p.other)] if p.other is not None else
               writer.to_dict(p.value)}
        return GraphSONUtil.typed_value("P", out)


class TextPSerializer(_GraphSONTypeIO):
    python_type = TextP

    @classmethod
    def dictify(cls, p, writer):
        out = {"predicate": p.operator,
               "value": [writer.to_dict(p.value), writer.to_dict(p.other)] if p.other is not None else
               writer.to_dict(p.value)}
        return GraphSONUtil.typed_value("TextP", out)


class BindingSerializer(_GraphSONTypeIO):
    python_type = Binding

    @classmethod
    def dictify(cls, binding, writer):
        out = {"key": binding.key,
               "value": writer.to_dict(binding.value)}
        return GraphSONUtil.typed_value("Binding", out)


class LambdaSerializer(_GraphSONTypeIO):
    python_type = FunctionType

    @classmethod
    def dictify(cls, lambda_object, writer):
        lambda_result = lambda_object()
        script = lambda_result if isinstance(lambda_result, str) else lambda_result[0]
        language = statics.default_lambda_language if isinstance(lambda_result, str) else lambda_result[1]
        out = {"script": script,
               "language": language}
        if language == "gremlin-groovy" and "->" in script:
            # if the user has explicitly added parameters to the groovy closure then we can easily detect one or two
            # arg lambdas - if we can't detect 1 or 2 then we just go with "unknown"
            args = script[0:script.find("->")]
            out["arguments"] = 2 if "," in args else 1
        else:
            out["arguments"] = -1

        return GraphSONUtil.typed_value("Lambda", out)


class TypeSerializer(_GraphSONTypeIO):
    python_type = TypeType

    @classmethod
    def dictify(cls, typ, writer):
        return writer.to_dict(typ())


class UUIDIO(_GraphSONTypeIO):
    python_type = uuid.UUID
    graphson_type = "g:UUID"
    graphson_base_type = "UUID"

    @classmethod
    def dictify(cls, obj, writer):
        return GraphSONUtil.typed_value(cls.graphson_base_type, str(obj))

    @classmethod
    def objectify(cls, d, reader):
        return cls.python_type(d)


class DateIO(_GraphSONTypeIO):
    graphson_type = "g:Date"
    graphson_base_type = "Date"

    @classmethod
    def dictify(cls, obj, writer):
        try:
            timestamp_seconds = calendar.timegm(obj.utctimetuple())
            pts = timestamp_seconds * 1e3 + getattr(obj, 'microsecond', 0) / 1e3
        except AttributeError:
            pts = calendar.timegm(obj.timetuple()) * 1e3

        ts = int(round(pts))
        return GraphSONUtil.typed_value(cls.graphson_base_type, ts)

    @classmethod
    def objectify(cls, ts, reader):
        # Python timestamp expects seconds
        return datetime.datetime.utcfromtimestamp(ts / 1000.0)


class OffsetDateTimeIO(_GraphSONTypeIO):
    python_type = datetime.datetime
    graphson_type = "gx:OffsetDateTime"
    graphson_base_type = "OffsetDateTime"

    @classmethod
    def dictify(cls, obj, writer):
        if obj.tzinfo is None:
            return DateIO.dictify(obj, writer)
        return GraphSONUtil.typed_value(cls.graphson_base_type, obj.isoformat(), "gx")

    @classmethod
    def objectify(cls, dt, reader):
        # specially handling as python isoformat does not support zulu until 3.11
        dt_iso = dt[:-1] + '+00:00' if dt.endswith('Z') else dt
        return datetime.datetime.fromisoformat(dt_iso)


# Based on current implementation, this class must always be declared before FloatIO.
# Seems pretty fragile for future maintainers. Maybe look into this.
class TimestampIO(_GraphSONTypeIO):
    """A timestamp in Python is type float"""
    python_type = statics.timestamp
    graphson_type = "g:Timestamp"
    graphson_base_type = "Timestamp"

    @classmethod
    def dictify(cls, obj, writer):
        # Java timestamp expects milliseconds integer
        # Have to use int because of legacy Python
        ts = int(round(obj * 1000))
        return GraphSONUtil.typed_value(cls.graphson_base_type, ts)

    @classmethod
    def objectify(cls, ts, reader):
        # Python timestamp expects seconds
        return cls.python_type(ts / 1000.0)


class _NumberIO(_GraphSONTypeIO):
    @classmethod
    def dictify(cls, n, writer):
        if isinstance(n, bool):  # because isinstance(False, int) and isinstance(True, int)
            return n
        return GraphSONUtil.typed_value(cls.graphson_base_type, n)

    @classmethod
    def objectify(cls, v, _):
        return cls.python_type(v)
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -