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 decimal import Decimal

from . 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 and
    casting Decimal instances as strings.
    """
    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
    elif isinstance(val, Decimal):
        return str(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(obj, **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, or Decimal values
    and we 'fix' it before re-encoding.
    """
    try:
        dumped = json.dumps(obj, allow_nan=False, **kwargs)
    except (ValueError, TypeError):
        obj = swap_inf_nan(obj)
        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