[functions][python] support both python 2 and 3
*Motivation*
base64 is non compatible between python2 and 3
*Changes*
Provides a python 2/3 compatible method
@@ -52,6 +52,20 @@ InternalMessage = namedtuple('InternalMessage', 'message topic serde consumer')
InternalQuitMessage = namedtuple('InternalQuitMessage', 'quit')
DEFAULT_SERIALIZER = "serde.IdentitySerDe"
+PY3 = sys.version_info[0] >= 3
+
+def base64ify(bytes_or_str):
+ if PY3 and isinstance(bytes_or_str, str):
+ input_bytes = bytes_or_str.encode('utf8')
+ else:
+ input_bytes = bytes_or_str
+
+ output_bytes = base64.urlsafe_b64encode(input_bytes)
+ if PY3:
+ return output_bytes.decode('ascii')
+ else:
+ return output_bytes
+
# We keep track of the following metrics
class Stats(object):
def __init__(self):
@@ -284,7 +298,7 @@ class PythonInstance(object):
self.current_stats.nserialization_exceptions += 1
self.total_stats.nserialization_exceptions += 1
if output_bytes is not None:
- props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64.b64encode(msg.message.message_id().serialize())}
+ props = {"__pfn_input_topic__" : str(msg.topic), "__pfn_input_msg_id__" : base64ify(msg.message.message_id().serialize())}
try:
self.producer.send_async(output_bytes, partial(self.done_producing, msg.consumer, msg.message), properties=props)
except Exception as e:
