sgr_commhandler.utils.jmespath_mapping

Provides an SGr-specific implementation that transforms JSON data using JMESpath mappings.

Classes

RecordKey

Implements a key with indices. Ported from Java.

Functions

map_json_response(response, mappings)

Converts the structure of a JSON string using JMESpath mappings.

_map_to_flat_list(json_str, keyword_map)

_enhance_with_namings(flat_representation, names)

_flatten_named_records(cur_key, values, names)

_parse_json_tree(node, parent_data, iteration, keyword_map)

_get_keywords_for_iteration(iteration, keyword_map)

_determine_required_iterations(keyword_map)

_get_number_of_elements(node, parent_idx, keyword, ...)

_process_child_elements(node, iteration, record_map, ...)

_add_child_element(node, record_map, key, kw, iteration)

_get_first_level_elements(keywords)

_get_second_level_elements(keywords)

_add_second_level_nodes(first_level_node, ...)

_build_json_node(keyword_map, flat_data_records)

Module Contents

class sgr_commhandler.utils.jmespath_mapping.RecordKey

Bases: object

Implements a key with indices. Ported from Java.

_indices: list[str] = []
add(idx)
Parameters:

idx (int)

index(iteration)
Parameters:

iteration (int)

Return type:

int

key()
Return type:

str

__eq__(other)
__hash__()
__str__()
Return type:

str

__repr__()
Return type:

str

sgr_commhandler.utils.jmespath_mapping.map_json_response(response, mappings)

Converts the structure of a JSON string using JMESpath mappings.

Parameters:

  • response (str) – the JSON string to convert

  • mappings (list[JmespathMappingRecord]) – the data point’s configured mappings

Returns:

the mapped JSON structure

Return type:

Any

sgr_commhandler.utils.jmespath_mapping._map_to_flat_list(json_str, keyword_map)
Parameters:

  • json_str (str)

  • keyword_map (dict[str, str])

Return type:

Optional[dict[RecordKey, dict[str, Any]]]

sgr_commhandler.utils.jmespath_mapping._enhance_with_namings(flat_representation, names)
Parameters:

  • flat_representation (dict[RecordKey, dict[str, Any]])

  • names (dict[str, str])

Return type:

dict[int, dict[str, Any]]

sgr_commhandler.utils.jmespath_mapping._flatten_named_records(cur_key, values, names)
Parameters:

  • cur_key (int)

  • values (dict[str, Any])

  • names (dict[str, str])

Return type:

tuple[int, dict[int, dict[str, Any]]]

sgr_commhandler.utils.jmespath_mapping._parse_json_tree(node, parent_data, iteration, keyword_map)
Parameters:

  • node (Any)

  • parent_data (Optional[dict[RecordKey, dict[str, Any]]])

  • iteration (int)

  • keyword_map (dict[str, str])

Return type:

Optional[dict[RecordKey, dict[str, Any]]]

sgr_commhandler.utils.jmespath_mapping._get_keywords_for_iteration(iteration, keyword_map)
Parameters:

  • iteration (int)

  • keyword_map (dict[str, str])

Return type:

list[tuple[str, str]]

sgr_commhandler.utils.jmespath_mapping._determine_required_iterations(keyword_map)
Parameters:

keyword_map (dict[str, str])

Return type:

int

sgr_commhandler.utils.jmespath_mapping._get_number_of_elements(node, parent_idx, keyword, iteration)
Parameters:

  • node (Any)

  • parent_idx (int)

  • keyword (tuple[str, str])

  • iteration (int)

Return type:

int

sgr_commhandler.utils.jmespath_mapping._process_child_elements(node, iteration, record_map, keywords, parent_idx, parent_rec)
Parameters:

  • node (Any)

  • iteration (int)

  • record_map (dict[RecordKey, dict[str, Any]])

  • keywords (list[tuple[str, str]])

  • parent_idx (int)

  • parent_rec (Optional[tuple[RecordKey, dict[str, Any]]])

sgr_commhandler.utils.jmespath_mapping._add_child_element(node, record_map, key, kw, iteration)
Parameters:

  • node (Any)

  • record_map (dict[RecordKey, dict[str, Any]])

  • key (RecordKey)

  • kw (tuple[str, str])

  • iteration (int)

sgr_commhandler.utils.jmespath_mapping._get_first_level_elements(keywords)
Parameters:

keywords (list[tuple[str, str]])

Return type:

dict[str, str]

sgr_commhandler.utils.jmespath_mapping._get_second_level_elements(keywords)
Parameters:

keywords (list[tuple[str, str]])

Return type:

dict[str, dict[str, str]]

sgr_commhandler.utils.jmespath_mapping._add_second_level_nodes(first_level_node, flat_records_belonging_to_group, keyword_map)
Parameters:

  • first_level_node (dict)

  • flat_records_belonging_to_group (list[dict[str, Any]])

  • keyword_map (dict[str, str])

sgr_commhandler.utils.jmespath_mapping._build_json_node(keyword_map, flat_data_records)
Parameters:

  • keyword_map (dict[str, str])

  • flat_data_records (list[dict[str, Any]])

Return type:

Any