flink-python/pyflink/common/serializer.py

################################################################################
#  Licensed to the Apache Software Foundation (ASF) under one
#  or more contributor license agreements.  See the NOTICE file
#  distributed with this work for additional information
#  regarding copyright ownership.  The ASF licenses this file
#  to you under the Apache License, Version 2.0 (the
#  "License"); you may not use this file except in compliance
#  with the License.  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
# limitations under the License.
################################################################################

from abc import abstractmethod, ABC
from io import BytesIO
from typing import TypeVar, Generic

T = TypeVar('T')

__all__ = ['TypeSerializer']


class TypeSerializer(ABC, Generic[T]):
    """
    This interface describes the methods that are required for a data type to be handled by the
    Flink runtime. Specifically, this interface contains the serialization and deserialization
    methods.
    """

    def __eq__(self, other):
        return isinstance(other, self.__class__) and self.__dict__ == other.__dict__

    def __ne__(self, other):
        return not self.__eq__(other)

    def __repr__(self):
        return "%s()" % self.__class__.__name__

    def __hash__(self):
        return hash(str(self))

    @abstractmethod
    def serialize(self, element: T, stream: BytesIO) -> None:
        """
        Serializes an element to the output stream.
        """
        pass

    @abstractmethod
    def deserialize(self, stream: BytesIO) -> T:
        """
        Returns a deserialized element from the input stream.
        """
        pass

    def _get_coder(self):
        serialize_func = self.serialize
        deserialize_func = self.deserialize

        class CoderAdapter(object):
            def get_impl(self):
                return CoderAdapterIml()

        class CoderAdapterIml(object):

            def encode_nested(self, element):
                bytes_io = BytesIO()
                serialize_func(element, bytes_io)
                return bytes_io.getvalue()

            def decode_nested(self, bytes_data):
                bytes_io = BytesIO(bytes_data)
                return deserialize_func(bytes_io)

        return CoderAdapter()


void = b''


class VoidNamespaceSerializer(TypeSerializer[bytes]):

    def serialize(self, element: bytes, stream: BytesIO) -> None:
        pass

    def deserialize(self, stream: BytesIO) -> bytes:
        return void