Dvl/klaus/webadm readable msg

pypeman/channels.py

@@ -12,6 +12,8 @@
 from pypeman.helpers.sleeper import Sleeper
 
 
+from pypeman.helpers.serializers import JsonableEncoder
+
 logger = logging.getLogger(__name__)
 
 # List all channel registered
@@ -469,6 +471,27 @@ def graph_dot(self, end=''):
         for end, sub in after:
             sub.graph_dot(end=end)
 
+    def jsonable_msg_info_for_admin(self, msg):
+        """ creates jsonable info representing the message for admin interfaces.
+            (Perhaps there's a better method name?)
+        """
+
+        encoder = JsonableEncoder()
+
+        result = {}
+        result['timestamp'] = msg.timestamp.strftime(message.DATE_FORMAT)
+        result['uuid'] = msg.uuid
+        result['payload'] = encoder.encode(msg.payload)
+        result['meta'] = msg.meta
+        result['ctx'] = {}
+
+        for k, ctx_msg in msg.ctx.items():
+            result['ctx'][k] = {}
+            result['ctx'][k]['payload'] = encoder.encode(ctx_msg['payload'])
+            result['ctx'][k]['meta'] = dict(ctx_msg['meta'])
+
+        return result
+
     def __str__(self):
         return "<chan: %s>" % self.name
 

pypeman/helpers/serializers.py

@@ -0,0 +1,100 @@
+"""
+serializers, that might be helpful for pypeman projects
+"""
+
+import json
+import logging
+import pickle
+
+from base64 import b64decode
+from base64 import b64encode
+
+
+logger = logging.getLogger(__name__)
+
+
+# attempt to create serializers allowing to encode objects for non
+# Python applications.
+class JsonableEncoder:
+    """ Initial simple encoder/decoder for objects.
+        returns:  a dict with type and value
+            type: 'asis' or 'b64', 'repr'
+            value:
+              - the object itself if it can be json encoded (basetype or
+                struct of base types)
+              - a b64 encoded string if obj is of type bytes and not ASCII
+              (- a slightly transformed object, that can be json encoded
+                    (To be implemented))
+              - a repr string of the object (which can normally not be decoded)
+    """
+    def encode(self, obj):
+        logger.debug("try to encode type %s (%r)", type(obj), obj)
+        result = {
+            'type': 'asis',
+            'value': obj,
+            'repr': '',
+            }
+        if isinstance(obj, bytes):
+            try:
+                result['type'] = "bytes"
+                result['value'] = obj.decode('ascii')  # OR UTF-8?
+                return result
+            except UnicodeDecodeError:
+                pass
+            try:
+                result['type'] = "utf8bytes"
+                # TODO could merge asis and utf8
+                result['value'] = obj.decode('utf-8')
+                return result
+            except UnicodeDecodeError:
+                result['type'] = "b64"
+                result['value'] = b64encode(obj).decode()
+                result['repr'] = obj.decode('utf-8', 'ignore')
+                return result
+        try:
+            # logger.info("try to dump %s", repr(obj))
+            json_str = json.dumps(obj)  # noqa this line will except if not jsonable
+            # logger.info("can be dumped as json %s", json_str)
+            return result
+        except TypeError:
+            pass
+        result['type'] = "repr"
+        # TODO: might add a base64 pickle for the repr case
+        result['value'] = None
+        result['repr'] = repr(obj)
+        return result
+
+    def decode(self, encoded_obj):
+        enc_type = encoded_obj['type']
+        data = encoded_obj['value']
+        if enc_type == 'asis':
+            return data
+        elif enc_type in ('bytes', 'utf8bytes'):
+            return data.encode('utf-8')
+        elif enc_type == 'b64':
+            return b64decode(data.encode('ascii'))
+
+
+class B64PickleEncoder:
+    """ Initial simple encoder for objects.
+            objects are encoded, such, that they can be stored / transfered as
+            an ASCII string.
+
+            drawback: the encoded object can only be decoded if the
+                application handles pickle. This limits this decoding mostly
+                to python3 applications.
+    """
+
+    def encode(self, obj):
+        """ the encoding function
+            :param obj: the object to be encoded
+            returns: a b64 encoded string of the pickled bytes of the passed
+                     object
+        """
+        return b64encode(pickle.dumps(obj)).decode('ascii')
+
+    def decode(self, encoded_obj):
+        """ the decoding function
+            :param encoded_obj: the encoded_object to be decoded
+        """
+        return pickle.loads(b64decode(encoded_obj.encode('ascii')))

pypeman/message.py

@@ -3,14 +3,16 @@
 import uuid
 import copy
 from uuid import UUID
-import pickle
 import json
-import base64
 import logging
 
+from pypeman.helpers.serializers import B64PickleEncoder
+
 default_logger = logging.getLogger(__name__)
 
 DATE_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
+DEFAULT_ENCODER_CLS = B64PickleEncoder
+
 
 
 class Message():
@@ -88,51 +90,79 @@ def add_context(self, key, msg):
             payload=copy.deepcopy(msg.payload),
         )
 
-    def to_dict(self):
+    def to_dict(self, payload_encoder=None):
         """
-        Convert the current message object to a dict. Payload is pickled.
+        Convert the current message object to a dict, that can be converted
+            to a json.
+        Therefore the payload and all the context is pickled and base64 encoded.
+
+        Warning: the returned dict cannot be converted to json if meta contains
+            objects, that cannot be dumped as json.
+
+        :param encoder: Encoder object that will be used to encode the payload
+            if set to None B64PickleEncoder() will be used, which just pickles an object
+            and encodes with base64.
 
         :return: A dict with an equivalent of message
         """
+
+        payload_encoder = (payload_encoder if payload_encoder
+                           else DEFAULT_ENCODER_CLS())
+
+        encode = payload_encoder.encode
         result = {}
         result['timestamp'] = self.timestamp.strftime(DATE_FORMAT)
         result['uuid'] = self.uuid
-        result['payload'] = base64.b64encode(pickle.dumps(self.payload)).decode('ascii')
+        result['payload'] = encode(self.payload)
         result['meta'] = self.meta
         result['ctx'] = {}
 
         for k, ctx_msg in self.ctx.items():
             result['ctx'][k] = {}
-            result['ctx'][k]['payload'] = base64.b64encode(pickle.dumps(ctx_msg['payload'])).decode('ascii')
+            result['ctx'][k]['payload'] = encode(ctx_msg['payload'])
             result['ctx'][k]['meta'] = dict(ctx_msg['meta'])
 
         return result
 
-    def to_json(self):
+    def to_json(self, payload_encoder=None):
         """
         Create json string for current message.
 
+        :param encoder: Encoder object that will be used to encode the payload.
+            more info at documentation of to_dict()
+
         :return: a json string equivalent for message.
         """
-        return json.dumps(self.to_dict())
+        return json.dumps(self.to_dict(payload_encoder=payload_encoder))
 
     @staticmethod
-    def from_dict(data):
+    def from_dict(data, payload_encoder=None):
         """
-        Convert the input dict previously converted with `.as_dict()` method in Message object.
+        Converts an input dict previously converted with `.to_dict()` method
+            to a Message object.
 
         :param data: The input dict.
-        :return: The message message object correponding to given data.
+
+        :param encoder: Encoder object that will be used to decode the payload
+            if set to None B64PickleEncoder() will be used, which decodes an
+                object that was pickled and then base64 encoded.
+
+        :return: The message message object corresponding to given data.
         """
+        payload_encoder = (payload_encoder if payload_encoder
+                           else DEFAULT_ENCODER_CLS())
+
+        decode = payload_encoder.decode
+
         result = Message()
         result.timestamp = datetime.datetime.strptime(data['timestamp'], DATE_FORMAT)
         result.uuid = UUID(data['uuid']).hex
-        result.payload = pickle.loads(base64.b64decode(data['payload'].encode('ascii')))
+        result.payload = decode(data['payload'])
         result.meta = data['meta']
 
         for k, ctx_msg in data['ctx'].items():
             result.ctx[k] = {}
-            result.ctx[k]['payload'] = pickle.loads(base64.b64decode(ctx_msg['payload'].encode('ascii')))
+            result.ctx[k]['payload'] = decode(ctx_msg['payload'])
             result.ctx[k]['meta'] = dict(ctx_msg['meta'])
 
         return result

pypeman/msgstore.py

@@ -57,7 +57,7 @@ async def get(self, id):
 
     async def search(self, start=0, count=10, order_by='timestamp'):
         """
-        Return a list of message with store specific `id` and processed status.
+        Return a list of messages with store specific `id` and processed status.
 
         :param start: First element.
         :param count: Count of elements since first element.
@@ -252,6 +252,10 @@ async def get(self, id):
             raise IndexError
 
         with open(os.path.join(self.base_path, id), "rb") as f:
+            # TODO: we might implement an async version for huge files
+            # - use either https://github.com/Tinche/aiofiles
+            # - read chunks + add sleep(0) (not good for blocking network file systems)
+
             msg = Message.from_json(f.read().decode('utf-8'))
             return {'id': id, 'state': await self.get_message_state(id), 'message': msg}
 

pypeman/remoteadmin.py

@@ -141,19 +141,34 @@ async def stop_channel(self, channel):
             'status': channels.BaseChannel.status_id_to_str(chan.status)
         }
 
-    async def list_msg(self, channel, start=0, count=10, order_by='timestamp'):
+    async def list_msg(self, channel, start=0, count=10, order_by='timestamp', mk_b64pickle=False):
         """
         List first `count` messages from message store of specified channel.
 
         :params channel: The channel name.
+        :param mk_b64pickle: if True (yield payload / ctx fields as b64 encoded pickles)
+                if False a a dict with a field type, field value and a field repr
+                will be created.
+                Field types depend on the channel's jsonable_msg_info_for_admin() method
+                typical values are:
+                    "asis": value is the object 'asis', as the object could be pickled
+                    "bytes": bytes just as ASCII string
+                    "utf8bytes": bytes as UTF8 string
+                    "b64":  base 64 encoded bytes
+                    "repr": a repr() string of the object is returned
+                    "b64pickle": all fields b64 encoded pickles of the related python object
         """
         chan = self.get_channel(channel)
 
         messages = await chan.message_store.search(start=start, count=count, order_by=order_by)
 
         for res in messages:
             res['timestamp'] = res['message'].timestamp_str()
-            res['message'] = res['message'].to_json()
+            if mk_b64pickle:
+                res['message'] = res['message'].to_json()
+            else:
+                msg = res['message']
+                res['message'] = chan.jsonable_msg_info_for_admin(msg)
 
         return {'messages': messages, 'total': await chan.message_store.total()}
 
@@ -180,7 +195,7 @@ async def push_msg(self, channel, text):
         Push a message in the channel.
 
         :params channel: The channel name.
-        :params msg_ids: The text added to the payload.
+        :params text: The text added to the payload.
         """
         chan = self.get_channel(channel)
         msg = message.Message(payload=text)
@@ -248,7 +263,8 @@ def stop(self, channel):
         """
         return self.send_command('stop_channel', [channel])
 
-    def list_msg(self, channel, start=0, count=10, order_by='timestamp'):
+    def list_msg(self, channel, start=0, count=10, order_by='timestamp',
+                 mk_b64pickle=True):
         """
         List first 10 messages on specified channel from remote instance.
 
@@ -258,7 +274,8 @@ def list_msg(self, channel, start=0, count=10, order_by='timestamp'):
         :params order_by: Message order. only 'timestamp' and '-timestamp' handled for now.
         :returns: list of message with status.
         """
-        result = self.send_command('list_msg', [channel, start, count, order_by])
+        result = self.send_command('list_msg', [channel, start, count, 
+                                   order_by, mk_b64pickle])
 
         for m in result['messages']:
             m['message'] = message.Message.from_json(m['message'])