Python simplejson 模块，JSONEncoder() 实例源码

def __init__(self, list_domain, list_key, list_name, list_description,
                 zk_hosts, aws_keyfile, s3_bucket, s3_endpoint="s3.amazonaws.com",
                 encoder_cls=json.JSONEncoder, decoder_cls=json.JSONDecoder,
                 update_callback=None, force_config_update=None):

        kwargs = {}
        if force_config_update is not None:
            kwargs['force_config_update'] = force_config_update

        super(ManagedJsonSerializableDataConfig, self).__init__(
            list_domain, list_key, list_name, list_description, zk_hosts,
            aws_keyfile, s3_bucket, s3_endpoint=s3_endpoint, **kwargs)

        self.encoder_cls = encoder_cls
        self.decoder_cls = decoder_cls
        self.update_callback = None
        if update_callback:
            self.set_update_callback(update_callback)

def default(self, obj):
                """Required routine that overrides the default base
                class version.  This routine must serialize 'obj' when
                attempting to save 'obj' json format."""

                if isinstance(obj, (pkg.fmri.PkgFmri,
                    pkg.client.linkedimage.common.LinkedImageName)):
                        return str(obj)

                if isinstance(obj, pkgplan.PkgPlan):
                        return obj.getstate()

                if isinstance(obj, (set, frozenset)):
                        return list(obj)

                return json.JSONEncoder.default(self, obj)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def encode(self, obj):
        if isinstance(obj, float):
            return format(obj, '.6f')
        return json.JSONEncoder.encode(self, obj)

def __init__(self, *args, **kwargs):
        super(JSONEncoder, self).__init__(*args, **kwargs)
        # Force to use our custom decimal with simplejson
        self.use_decimal = False

def default(self, obj):
        marshaller = self.serializers.get(type(obj),
            super(JSONEncoder, self).default)
        return marshaller(obj)

def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
        """Dispatches an JSON-RPC method from marshalled (JSON) data.

        JSON-RPC methods are dispatched from the marshalled (JSON) data
        using the _dispatch method and the result is returned as
        marshalled data. For backwards compatibility, a dispatch
        function can be provided as an argument (see comment in
        SimpleJSONRPCRequestHandler.do_POST) but overriding the
        existing method through subclassing is the prefered means
        of changing method dispatch behavior.
        """
        rawreq = json.loads(data, object_hook=JSONDecoder())

        req_id = rawreq.get('id', 0)
        method = rawreq['method']
        params = rawreq.get('params', [])

        response = {'id': req_id}

        try:
            # generate response
            if dispatch_method is not None:
                response['result'] = dispatch_method(method, params)
            else:
                response['result'] = self._dispatch(method, params)
        except (UserError, UserWarning, NotLogged,
                ConcurrencyException), exception:
            response['error'] = exception.args
        except Exception:
            tb_s = ''.join(traceback.format_exception(*sys.exc_info()))
            for path in sys.path:
                tb_s = tb_s.replace(path, '')
            # report exception back to server
            response['error'] = (str(sys.exc_value), tb_s)

        return json.dumps(response, cls=JSONEncoder)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def _encoder_default_args(kw):
    """Shape default arguments for encoding functions."""

    # manual override of the preferred coding with `exact=False`
    if kw.pop('exact', getattr(_local, 'coding', CODING_DEFAULT) == EXACT):
        # settings necessary for the "exact coding"
        kw.update({
            'default': _json_default_exact,
            'use_decimal': False,           # don't encode `Decimal` as JSON's `Number`
            'tuple_as_array': False,        # don't encode `tuple` as `Array`
            'namedtuple_as_object': False   # don't call `_asdict` on `namedtuple`
        })
    else:
        # settings for the "compatibility coding"
        kw.update({
            'default': _json_default_compat,
            'ignore_nan': True      # be compliant with the ECMA-262 specification:
                                    # serialize nan/inf as null
        })

    # NOTE: if called from ``simplejson.dumps()`` with ``cls=JSONEncoder``,
    # we will receive all kw set to simplejson defaults -- and our defaults for
    # ``separators`` and ``for_json`` will not be applied. In contrast, they
    # are applied when called from ``jsonplus.dumps()``, unless user explicitly
    # sets some of those.
    # This causes inconsistent behaviour between ``dumps()`` and ``JSONEncoder()``.

    # prefer compact json repr
    kw.setdefault('separators', (',', ':'))

    # allow objects to provide json serialization on its behalf
    kw.setdefault('for_json', True)

def __init__(self, **kw):
        """Constructor for simplejson.JSONEncoder, with defaults overriden
        for jsonplus.
        """
        _encoder_default_args(kw)
        super(JSONEncoder, self).__init__(**kw)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def default(self, obj):
        if isinstance(obj, np.ndarray):
                return obj.tolist()
        return json.JSONEncoder.default(self, obj)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def default(self, o):
        if isinstance(o, datetime):
            return o.isoformat()

        elif isinstance(o, bytes):
            return o.decode('utf-8')

        elif hasattr(o, '__table__'):  # SQLAlchemy model
            return o.to_dict()

        elif o is int:
            return 'int'

        elif o is float:
            return 'float'

        elif type(o).__name__ == 'ndarray': # avoid numpy import
            return o.tolist()

        elif type(o).__name__ == 'DataFrame':  # avoid pandas import
            o.columns = o.columns.droplevel('channel')  # flatten MultiIndex
            return o.to_dict(orient='index')

        elif type(o) is type and o in data_types:
            return data_types[o]

        return json.JSONEncoder.default(self, o)

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def default(self, obj):
        try:
            return super(JSONEncoder, self).default(obj)
        except TypeError:
            reducer = getattr(obj, '__to_json__', None)
            if reducer:
                return reducer()
            raise

def install_json():
    json._default_encoder = JSONEncoder()
    json._default_decoder.object_hook = decode_hook
    try:
        from kombu.utils import json as kombujson
    except ImportError:
        pass
    else:
        kombujson._default_encoder = JSONEncoder

def default(self, obj):
        if isinstance(obj, DottedCollection):
            return obj.store
        else:
            return json.JSONEncoder.default(obj)

def iter_dumps(obj, **kw):
    return ''.join(json.JSONEncoder(**kw).iterencode(obj))

def test_encoding1(self):
        encoder = json.JSONEncoder(encoding='utf-8')
        u = u'\N{GREEK SMALL LETTER ALPHA}\N{GREEK CAPITAL LETTER OMEGA}'
        s = u.encode('utf-8')
        ju = encoder.encode(u)
        js = encoder.encode(s)
        self.assertEqual(ju, js)

def default(self, o):
        if o is JSONTestObject:
            if self.recurse:
                return [JSONTestObject]
            else:
                return 'JSONTestObject'
        return json.JSONEncoder.default(o)

def default(self, obj):
    """Arranges entity/entity group access counts by their kind.

    Args:
      obj: an object whose JSON encoding is desired.
    Returns:
      JSON encoding of obj.
    """
    if not isinstance(obj, RPCSummary):
      return json.JSONEncoder.default(self, obj)
    return obj.__dict__

def default(self, obj):
        if hasattr(obj, 'to_dict'):
            return maybe_call(obj.to_dict)
        elif hasattr(obj, 'to_list'):
            return maybe_call(obj.to_list)
        elif isinstance(obj, types.GeneratorType):
            return list(obj)
        return super(JSONEncoder, self).default(obj)

def __init__(self, *args, **kwargs):
        super(JSONEncoder, self).__init__(*args, **kwargs)
        # Force to use our custom decimal with simplejson
        self.use_decimal = False

def default(self, obj):
        marshaller = self.serializers.get(
            type(obj),
            super(JSONEncoder, self).default
        )
        return marshaller(obj)

def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return _encode_time(obj)
        return simplejson.JSONEncoder.default(self, obj)

def default(self, obj):
        if isinstance(obj, (datetime.datetime, datetime.date)):
            return obj.isoformat()
        elif isinstance(obj, ObjectId):
            return str(obj)
        return json.JSONEncoder.default(self, obj)

def default(self, obj):
        if isinstance(obj, datetime):
            return obj.strftime('%Y-%m-%d %H:%M:%S')
        elif isinstance(obj, date):
            return obj.strftime('%Y-%m-%d')
        elif isinstance(obj, ObjectId):
            return str(obj)
        return json.JSONEncoder.default(self, obj)

def _dump(self, obj, fp, skipkeys=False, ensure_ascii=True,
            check_circular=True, allow_nan=True, cls=json.JSONEncoder,
            indent=None, separators=None, encoding='utf-8', default=None, **kw):
                iterable = cls(skipkeys=skipkeys, ensure_ascii=ensure_ascii,
                    check_circular=check_circular, allow_nan=allow_nan,
                    indent=indent, separators=separators, encoding=encoding,
                    default=default, **kw).iterencode(obj,
                    _one_shot=self.__single_pass)
                fp.writelines(misc.force_bytes(i) for i in iterable)