Protobuf --> Json and back

pb2json.py

""" 
:module pbf2json:

--------------------------------------------------------------------------------
License:    Public Domain with no guarantees made 
Date:       12/28/2012 

http://cjhanks.name 
--------------------------------------------------------------------------------
        
Utilities to convert a protobuf instance into a JSON serial.

There didn't seem to be a great reason to add support for a JSON --> PB, so it 
wasn't implemented. 
"""

import json 
from   google.protobuf.descriptor import FieldDescriptor as gfd 


def __devel_err(_arg_):
    """ raises a ValueError to deal with gfd.TYPE_MESSAGE being called 
    from cast_map improperly
    """
    raise ValueError('development / casting error')

#
# Cast CPP types --> PyType 
#
# pylint: disable-msg=C0103
__cast_map = {
    gfd.TYPE_DOUBLE     : float,
    gfd.TYPE_FLOAT      : float,
    gfd.TYPE_INT64      : long,
    gfd.TYPE_UINT64     : long,
    gfd.TYPE_INT32      : int,
    gfd.TYPE_FIXED64    : float,
    gfd.TYPE_FIXED32    : float,
    gfd.TYPE_BOOL       : bool,
    gfd.TYPE_STRING     : unicode,
    gfd.TYPE_MESSAGE    : __devel_err, 
    gfd.TYPE_BYTES      : lambda x: x.encode('string_escape'),
    gfd.TYPE_UINT32     : int,
    gfd.TYPE_ENUM       : int,
    gfd.TYPE_SFIXED32   : float,
    gfd.TYPE_SFIXED64   : float,
    gfd.TYPE_SINT32     : int,
    gfd.TYPE_SINT64     : long,
}
# pylint: enable-msg=C0103


def pb_to_dict(pbf):
    """
    Recursive function which fills in a dictionary with all of the defined 
    paramaters 

    :param pbf:  Protobuf instance to convert to dict() 
    :type  pbf:  google.protobuf.message 

    :return:    Dictionary model of protobuf  
    :type:      dict() 

    :raises RuntimeError:   When the protobuf is not initialized. 
    :raises ValueError:     When the protobuf 
    """
    if not pbf.IsInitialized():
        raise RuntimeError('Protobuf not valid') 
    
    ret = dict() 

    for field, value in pbf.ListFields():
        if   field.type == gfd.TYPE_MESSAGE: 
            cast = pb_to_json 
        elif field.type in __cast_map:
            cast = __cast_map[field.type] 
        else:
            raise ValueError('Unknkown type %s' % field.name)

        if field.label == gfd.LABEL_REPEATED:
            ret[field.name] = list() 
            
            for its in value:
                ret[field.name].append(cast(its))
        else:
            ret[field.name] = cast(value) 

    return ret  


def pb_to_json(pbf):
    """
    Creates a JSON string from an existing protobuf 
         
    :param pbf:  Protobuf instance to convert to dict() 
    :type  pbf:  google.protobuf.message 
                   
    :return:    JSON string 
    :type:      str() 
                                           
    :raises RuntimeError:   When the protobuf is not initialized. 
    :raises ValueError:     When the protobuf contains an unknown data type
    """
    return json.dumps(pb_to_dict(pbf)) 


if __name__ == '__main__':
    from notification_pb2 import log_event 

    e = log_event() 
    e.version  = log_event.ver_00 
    e.sender   = 'frank'
    e.loglevel = 3 
    e.err_maj  = 1
    e.err_min  = 2
    e.data     = 'Ancillary data'
    print(pb_to_json(e))