Skip to content

Instantly share code, notes, and snippets.

@timfeirg
Last active March 25, 2018 06:52
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save timfeirg/ff24713be505e7332db77960d8bbcba4 to your computer and use it in GitHub Desktop.
Save timfeirg/ff24713be505e7332db77960d8bbcba4 to your computer and use it in GitHub Desktop.
the world's only json encoder that handles buffer message (or grpc message, I don't know)
import json
from datetime import datetime
from decimal import Decimal
from flask import Response
from functools import wraps
from google.protobuf.internal import api_implementation
if api_implementation.Type() == 'cpp':
from google.protobuf.pyext.cpp_message import GeneratedProtocolMessageType
from google.protobuf.pyext._message import ScalarMapContainer as ScalarMap
from google.protobuf.pyext._message import RepeatedScalarContainer as RepeatedScalarFieldContainer
from google.protobuf.pyext._message import RepeatedCompositeContainer as RepeatedCompositeFieldContainer
else:
from google.protobuf.internal.python_message import GeneratedProtocolMessageType
from google.protobuf.internal.containers import ScalarMap, RepeatedScalarFieldContainer, RepeatedCompositeFieldContainer
class VersatileEncoder(json.JSONEncoder):
def convert_grpc_types(self, obj):
if isinstance(obj, ScalarMap):
return dict(obj)
if isinstance(obj, RepeatedScalarFieldContainer):
return list(obj)
if isinstance(type(obj), GeneratedProtocolMessageType):
return self.default(obj)
if isinstance(obj, RepeatedCompositeFieldContainer):
return [self.default(o) for o in obj]
return obj
def default(self, obj):
if hasattr(obj, 'to_dict'):
return obj.to_dict()
if isinstance(obj, datetime):
return obj.strftime('%Y-%m-%d %H:%M:%S')
if isinstance(obj, Decimal):
return float(obj)
if isinstance(obj, bytes):
return obj.decode('utf-8')
if isinstance(type(obj), GeneratedProtocolMessageType):
field_names = [field.name for field in obj.DESCRIPTOR.fields]
dic = {n: self.convert_grpc_types(getattr(obj, n)) for n in field_names}
# need this to identify what kind of message is this
dic['__class__'] = obj.__class__.__name__
return dic
return super(VersatileEncoder, self).default(obj)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment