json

get(dict_inst, key_str)

Return the value based on the key_str specified.

The following are equivalent::

obj['a']['b']['c'] <=> get(obj, 'a.b.c').value

provided that the keys a, b and c are valid.

Parameters:

Name	Type	Description	Default
dict_inst	Any	The dictionary instance to query.	required
key_str	str	A simple query to fetch from the dict_inst.	required

Returns:

Type	Description
OneOf[Issue, Any]	A Good if the value we want is available, otherwise it returns a
OneOf[Issue, Any]	Bad with the path that returned None.

Source code in m/core/json.py

def get(dict_inst: Any, key_str: str) -> OneOf[Issue, Any]:
    """Return the value based on the `key_str` specified.

    The following are equivalent::

        obj['a']['b']['c'] <=> get(obj, 'a.b.c').value

    provided that the keys `a`, `b` and `c` are valid.

    Args:
        dict_inst: The dictionary instance to query.
        key_str: A simple query to fetch from the dict_inst.

    Returns:
        A `Good` if the value we want is available, otherwise it returns a
        `Bad` with the path that returned `None`.
    """
    keys = key_str.split('.')
    current = dict_inst
    for num, key in enumerate(keys):
        new_key: Union[str, int] = key
        with suppress(ValueError):
            new_key = int(key)
        try:
            current = current[new_key]
        except KeyError:
            pth = '.'.join(keys[:num + 1])
            return issue(f'`{pth}` path was not found')
        except Exception as ex:
            pth = '.'.join(keys[:num])
            if not isinstance(current, Mapping):
                context = pth or current
                return issue(f'`{context}` is not a dict')
            # catch unknown issue
            return issue(  # pragma: no cover
                f'{pth} resulted in an error',
                cause=ex,
            )
    return Good(current)

jsonq(dict_inst, separator, *key_str)

Stringify the values obtained from multi_get.

Parameters:

Name	Type	Description	Default
dict_inst	Mapping[str, Any]	The dictionary instance to query.	required
separator	str	A string to use to separate the results.	required
key_str	str	The queries to apply.	()

Returns:

Type	Description
OneOf[Issue, str]	A string separated by separator if successful or an Issue with
OneOf[Issue, str]	a description of the problems.

Source code in m/core/json.py

def jsonq(
    dict_inst: Map[str, Any],
    separator: str,
    *key_str: str,
) -> OneOf[Issue, str]:
    """Stringify the values obtained from `multi_get`.

    Args:
        dict_inst: The dictionary instance to query.
        separator: A string to use to separate the results.
        key_str: The queries to apply.

    Returns:
        A string separated by `separator` if successful or an Issue with
        a description of the problems.
    """
    return one_of(lambda: [
        separator.join(map(_to_str, key_values))
        for key_values in multi_get(dict_inst, *key_str)
    ])

multi_get(dict_inst, *keys)

Call get for every input specified by keys.

It collects the invalid keys and returns an Issue::

multi_get(obj, 'a', 'a.b', 'a.b.c')

Parameters:

Name	Type	Description	Default
dict_inst	object	The dictionary instance to query.	required
keys	str	The queries to apply.	()

Returns:

Type	Description
OneOf[Issue, list[Any]]	A Good with a list of the results if successful, otherwise a Bad
OneOf[Issue, list[Any]]	with the list of failures.

Source code in m/core/json.py

def multi_get(
    dict_inst: object,
    *keys: str,
) -> OneOf[Issue, list[Any]]:
    """Call `get` for every input specified by `keys`.

    It collects the invalid keys and returns an `Issue`::

        multi_get(obj, 'a', 'a.b', 'a.b.c')

    Args:
        dict_inst: The dictionary instance to query.
        keys: The queries to apply.

    Returns:
        A `Good` with a list of the results if successful, otherwise a `Bad`
        with the list of failures.
    """
    result_items = []
    failures = []
    for key in keys:
        res = get(dict_inst, key)
        if res.is_bad:
            failures.append(
                Issue(
                    message=f'key lookup failure: `{key}`',
                    cause=res.value,
                    include_traceback=False,
                ),
            )
        else:
            result_items.append(res.value)
    if failures:
        return issue(
            'multi_get key retrieval failure',
            context=[x.to_dict() for x in failures],
            include_traceback=False,
        )
    return Good(result_items)

parse_json(json_str, error_if_empty=False)

Parse a string as json.

Parameters:

Name	Type	Description	Default
json_str	str	The string to parse.	required
error_if_empty	bool	The json parser may throw an error if True.	False

Returns:

Type	Description
OneOf[Issue, Any]	A Good containing the parsed contents of the json string.

Source code in m/core/json.py

def parse_json(
    json_str: str,
    error_if_empty: bool = False,
) -> OneOf[Issue, Any]:
    """Parse a string as json.

    Args:
        json_str: The string to parse.
        error_if_empty: The json parser may throw an error if `True`.

    Returns:
        A `Good` containing the parsed contents of the json string.
    """
    empty = '' if error_if_empty else 'null'
    try:
        return Good(json.loads(json_str or empty))
    except Exception as ex:
        return issue('failed to parse the json data', cause=ex)

read_json(filename, error_if_empty=False)

Read a json object from a json file.

Parameters:

Name	Type	Description	Default
filename	str | None	The filename to read from, if None it reads from stdin.	required
error_if_empty	bool	The json parser may throw an error if True.	False

Returns:

Type	Description
OneOf[Issue, Any]	A Good containing the parsed contents of the json file.

Source code in m/core/json.py

def read_json(
    filename: str | None,
    error_if_empty: bool = False,
) -> OneOf[Issue, Any]:
    """Read a json object from a json file.

    Args:
        filename: The filename to read from, if `None` it reads from stdin.
        error_if_empty: The json parser may throw an error if `True`.

    Returns:
        A `Good` containing the parsed contents of the json file.
    """
    empty: str = '' if error_if_empty else 'null'
    context = {'filename': filename or 'SYS.STDIN'}
    return one_of(lambda: [
        json_data
        for json_str in rw.read_file(filename)
        for json_data in parse_json(json_str or empty, error_if_empty)
    ]).flat_map_bad(hone('failed to read json file', context=context))