Source code for galaxy.util.json

from __future__ import absolute_import

import collections
import copy
import json
import logging
import math
import random
import string

from six import iteritems, string_types

from ..util import unicodify

__all__ = ("safe_dumps", "validate_jsonrpc_request", "validate_jsonrpc_response", "jsonrpc_request", "jsonrpc_response")

log = logging.getLogger(__name__)

to_json_string = json.dumps
from_json_string = json.loads


def swap_inf_nan(val):
    """
    This takes an arbitrary object and preps it for jsonifying safely, templating Inf/NaN.
    """
    if isinstance(val, string_types):
        # basestring first, because it's a sequence and would otherwise get caught below.
        return val
    elif isinstance(val, collections.Sequence):
        return [swap_inf_nan(v) for v in val]
    elif isinstance(val, collections.Mapping):
        return dict([(swap_inf_nan(k), swap_inf_nan(v)) for (k, v) in iteritems(val)])
    elif isinstance(val, float):
        if math.isnan(val):
            return "__NaN__"
        elif val == float("inf"):
            return "__Infinity__"
        elif val == float("-inf"):
            return "__-Infinity__"
        else:
            return val
    else:
        return val


def safe_loads(arg):
    """
    This is a wrapper around loads that returns the parsed value instead of
    raising a value error. It also avoids autoconversion of non-iterables
    i.e numeric and boolean values.
    """
    try:
        loaded = json.loads(arg)
        if loaded is not None and not isinstance(loaded, collections.Iterable):
            loaded = arg
    except (TypeError, ValueError):
        loaded = arg
    return loaded


[docs]def safe_dumps(*args, **kwargs): """ This is a wrapper around dumps that encodes Infinity and NaN values. It's a fairly rare case (which will be low in request volume). Basically, we tell json.dumps to blow up if it encounters Infinity/NaN, and we 'fix' it before re-encoding. """ try: dumped = json.dumps(*args, allow_nan=False, **kwargs) except ValueError: obj = swap_inf_nan(copy.deepcopy(args[0])) dumped = json.dumps(obj, allow_nan=False, **kwargs) if kwargs.get('escape_closing_tags', True): return dumped.replace('</', '<\\/') return dumped
def safe_dumps_formatted(obj): """Attempt to format an object for display. If serialization fails, the object's string representation will be returned instead. """ try: return safe_dumps(obj, sort_keys=True, indent=4, separators=(',', ': ')) except TypeError: log.warning("JSON serialization failed for object: %s", str(obj)) return str(obj) # Methods for handling JSON-RPC
[docs]def validate_jsonrpc_request(request, regular_methods, notification_methods): try: request = json.loads(request) except Exception as e: return False, request, jsonrpc_response(id=None, error=dict(code=-32700, message='Parse error', data=unicodify(e))) try: assert 'jsonrpc' in request, \ 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Request object as per the JSON-RPC 2.0 Specification.' assert request['jsonrpc'] == '2.0', \ 'Requested JSON-RPC version "%s" != required version "2.0".' % request['jsonrpc'] assert 'method' in request, 'No "method" member was sent with the Request object' except AssertionError as e: return False, request, jsonrpc_response(request=request, error=dict(code=-32600, message='Invalid Request', data=unicodify(e))) try: assert request['method'] in (regular_methods + notification_methods) except AssertionError: return False, request, jsonrpc_response(request=request, error=dict(code=-32601, message='Method not found', data='Valid methods are: %s' % ', '.join(regular_methods + notification_methods))) try: if request['method'] in regular_methods: assert 'id' in request, 'No "id" member was sent with the Request object and the requested method "%s" is not a notification method' % request['method'] except AssertionError as e: return False, request, jsonrpc_response(request=request, error=dict(code=-32600, message='Invalid Request', data=unicodify(e))) return True, request, None
[docs]def validate_jsonrpc_response(response, id=None): try: response = json.loads(response) except Exception as e: log.error('Response was not valid JSON: %s', unicodify(e)) log.debug('Response was: %s', response) return False, response try: assert 'jsonrpc' in response, \ 'This server requires JSON-RPC 2.0 and no "jsonrpc" member was sent with the Response object as per the JSON-RPC 2.0 Specification.' assert ('result' in response or 'error' in response), \ 'Neither of "result" or "error" members were sent with the Response object.' if 'error' in response: assert int(response['error']['code']), \ 'The "code" member of the "error" object in the Response is missing or not an integer.' assert 'message' in response, \ 'The "message" member of the "error" object in the Response is missing.' except Exception: log.exception('Response was not valid JSON-RPC') log.debug('Response was: %s' % response) return False, response if id is not None: try: assert 'id' in response and response['id'] == id except Exception: log.error('The response id "%s" does not match the request id "%s"' % (response['id'], id)) return False, response return True, response
[docs]def jsonrpc_request(method, params=None, id=None, jsonrpc='2.0'): if method is None: log.error('jsonrpc_request(): "method" parameter cannot be None') return None request = dict(jsonrpc=jsonrpc, method=method) if params: request['params'] = params if id is not None and id is True: request['id'] = ''.join(random.choice(string.hexdigits) for i in range(16)) elif id is not None: request['id'] = id return request
[docs]def jsonrpc_response(request=None, id=None, result=None, error=None, jsonrpc='2.0'): if result: rval = dict(jsonrpc=jsonrpc, result=result) elif error: rval = dict(jsonrpc=jsonrpc, error=error) else: msg = 'jsonrpc_response() called with out a "result" or "error" parameter' log.error(msg) rval = dict(jsonrpc=jsonrpc, error=msg) if id is not None: rval['id'] = id elif request is not None and 'id' in request: rval['id'] = request['id'] return rval