Warning

This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.

Source code for galaxy.util.json

import copy
import json
import logging
import math
import random
import string
from collections.abc import (
    Iterable,
    Mapping,
    Sequence,
)


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, str):
        # basestring first, because it's a sequence and would otherwise get caught below.
        return val
    elif isinstance(val, Sequence):
        return [swap_inf_nan(v) for v in val]
    elif isinstance(val, Mapping):
        return {swap_inf_nan(k): swap_inf_nan(v) for (k, v) in val.items()}
    elif isinstance(val, float):
        if math.isnan(val):
            return "__NaN__"
        elif val == math.inf:
            return "__Infinity__"
        elif val == -math.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, Iterable):
            loaded = arg
    except (TypeError, ValueError):
        loaded = copy.deepcopy(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(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', \ f"Requested JSON-RPC version \"{request['jsonrpc']}\" != required version \"2.0\"." 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=f"Valid methods are: {', '.join(regular_methods + notification_methods)}")) try: if request['method'] in regular_methods: assert 'id' in request, f"No \"id\" member was sent with the Request object and the requested method \"{request['method']}\" is not a notification 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(f'Response was: {response}') return False, response if id is not None: try: assert 'id' in response and response['id'] == id except Exception: log.error(f"The response id \"{response['id']}\" does not match the request 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