1""" 2raven.utils.json 3~~~~~~~~~~~~~~~~~~~~~~~~ 4 5:copyright: (c) 2010-2012 by the Sentry Team, see AUTHORS for more details. 6:license: BSD, see LICENSE for more details. 7""" 8 9from __future__ import absolute_import 10 11import codecs 12import collections 13import datetime 14import uuid 15import json 16 17from .basic import is_namedtuple 18 19 20try: 21 JSONDecodeError = json.JSONDecodeError 22except AttributeError: 23 JSONDecodeError = ValueError 24 25 26class BetterJSONEncoder(json.JSONEncoder): 27 ENCODER_BY_TYPE = { 28 uuid.UUID: lambda o: o.hex, 29 datetime.datetime: lambda o: o.strftime('%Y-%m-%dT%H:%M:%SZ'), 30 set: list, 31 frozenset: list, 32 bytes: lambda o: o.decode('utf-8', errors='replace'), 33 collections.namedtuple: lambda o: o._asdict(), 34 } 35 36 def default(self, obj): 37 obj_type = type(obj) 38 if obj_type not in self.ENCODER_BY_TYPE and is_namedtuple(obj): 39 obj_type = collections.namedtuple 40 41 try: 42 encoder = self.ENCODER_BY_TYPE[obj_type] 43 except KeyError: 44 try: 45 return super(BetterJSONEncoder, self).default(obj) 46 except Exception: 47 try: 48 return repr(obj) 49 except Exception: 50 return object.__repr__(obj) 51 return encoder(obj) 52 53 54def better_decoder(data): 55 return data 56 57 58def dumps(value, **kwargs): 59 try: 60 return json.dumps(value, cls=BetterJSONEncoder, **kwargs) 61 except Exception: 62 kwargs['encoding'] = 'safe-utf-8' 63 return json.dumps(value, cls=BetterJSONEncoder, **kwargs) 64 65 66def loads(value, **kwargs): 67 return json.loads(value, object_hook=better_decoder) 68 69 70_utf8_encoder = codecs.getencoder('utf-8') 71 72 73def safe_encode(input, errors='backslashreplace'): 74 return _utf8_encoder(input, errors) 75 76 77_utf8_decoder = codecs.getdecoder('utf-8') 78 79 80def safe_decode(input, errors='replace'): 81 return _utf8_decoder(input, errors) 82 83 84class Codec(codecs.Codec): 85 86 def encode(self, input, errors='backslashreplace'): 87 return safe_encode(input, errors) 88 89 def decode(self, input, errors='replace'): 90 return safe_decode(input, errors) 91 92 93class IncrementalEncoder(codecs.IncrementalEncoder): 94 def encode(self, input, final=False): 95 return safe_encode(input, self.errors)[0] 96 97 98class IncrementalDecoder(codecs.IncrementalDecoder): 99 def decode(self, input, final=False): 100 return safe_decode(input, self.errors)[0] 101 102 103class StreamWriter(Codec, codecs.StreamWriter): 104 pass 105 106 107class StreamReader(Codec, codecs.StreamReader): 108 pass 109 110 111def getregentry(name): 112 if name == 'safe-utf-8': 113 return codecs.CodecInfo( 114 name=name, 115 encode=safe_encode, 116 decode=safe_decode, 117 incrementalencoder=IncrementalEncoder, 118 incrementaldecoder=IncrementalDecoder, 119 streamreader=StreamReader, 120 streamwriter=StreamWriter, 121 ) 122 123 124codecs.register(getregentry) 125