22
33import re
44import warnings
5- from typing import Any , Dict , List , Mapping , Optional , Union
5+ from collections .abc import Mapping
6+ from typing import Any , Dict , List , Optional , Union
67
78import jmespath
89
1617)
1718
1819
19- def extract_data_from_json (data : Union [Mapping , List ], path : str = "*" , exclude : Optional [List ] = None ) -> Any :
20+ def extract_data_from_json (
21+ data : Union [Mapping , List ],
22+ path : str = "*" ,
23+ exclude : Optional [List ] = None ,
24+ ) -> Any :
2025 """Return wanted data from outpdevice data based on the check path. See unit test for complete example.
2126
2227 Get the wanted values to be evaluated if JMESPath expression is defined,
@@ -35,12 +40,16 @@ def extract_data_from_json(data: Union[Mapping, List], path: str = "*", exclude:
3540 """
3641 if exclude and isinstance (data , (Dict , List )):
3742 if not isinstance (exclude , list ):
38- raise ValueError (f"Exclude list must be defined as a list. You have { type (exclude )} " )
43+ raise ValueError (
44+ f"Exclude list must be defined as a list. You have { type (exclude )} " ,
45+ )
3946 # exclude unwanted elements
4047 exclude_filter (data , exclude )
4148
4249 if not path :
43- warnings .warn ("JMSPath cannot be empty string or type 'None'. Path argument reverted to default value '*'" )
50+ warnings .warn (
51+ "JMSPath cannot be empty string or type 'None'. Path argument reverted to default value '*'" ,
52+ )
4453 path = "*"
4554
4655 if path == "*" :
@@ -50,7 +59,10 @@ def extract_data_from_json(data: Union[Mapping, List], path: str = "*", exclude:
5059 # Multi ref_key
5160 if len (re .findall (r"\$.*?\$" , path )) > 1 :
5261 clean_path = path .replace ("$" , "" )
53- values = jmespath .search (f"{ clean_path } { ' | []' * (path .count ('*' ) - 1 )} " , data )
62+ values = jmespath .search (
63+ f"{ clean_path } { ' | []' * (path .count ('*' ) - 1 )} " ,
64+ data ,
65+ )
5466 return keys_values_zipper (
5567 multi_reference_keys (path , data ),
5668 associate_key_of_my_value (clean_path , values ),
@@ -59,41 +71,62 @@ def extract_data_from_json(data: Union[Mapping, List], path: str = "*", exclude:
5971 values = jmespath .search (jmespath_value_parser (path ), data )
6072
6173 if values is None :
62- raise TypeError ("JMSPath returned 'None'. Please, verify your JMSPath regex." )
74+ raise TypeError (
75+ "JMSPath returned 'None'. Please, verify your JMSPath regex." ,
76+ )
6377
6478 # check for multi-nested lists
65- if any (isinstance (i , list ) for i in values ):
79+ if not isinstance ( values , ( str , int , float , bool )) and any (isinstance (i , list ) for i in values ):
6680 # process elements to check if lists should be flattened
6781 for element in values :
6882 for item in element :
6983 # raise if there is a dict, path must be more specific to extract data
7084 if isinstance (item , dict ):
7185 raise TypeError (
72- f'Must be list of lists i.e. [["Idle", 75759616], ["Idle", 75759620]]. You have "{ values } ".'
86+ f'Must be list of lists i.e. [["Idle", 75759616], ["Idle", 75759620]]. You have "{ values } ".' ,
7387 )
7488 if isinstance (item , list ):
75- values = flatten_list (values ) # flatten list and rewrite values
89+ values = flatten_list (
90+ values ,
91+ ) # flatten list and rewrite values
7692 break # items are the same, need to check only first to see if this is a nested list
7793
7894 # We need to get a list of reference keys - list of strings.
7995 # Based on the expression or data we might have different data types
8096 # therefore we need to normalize.
8197 if re .search (r"\$.*\$" , path ):
82- paired_key_value = associate_key_of_my_value (jmespath_value_parser (path ), values )
83- wanted_reference_keys = jmespath .search (jmespath_refkey_parser (path ), data )
98+ paired_key_value = associate_key_of_my_value (
99+ jmespath_value_parser (path ),
100+ values ,
101+ )
102+ wanted_reference_keys = jmespath .search (
103+ jmespath_refkey_parser (path ),
104+ data ,
105+ )
84106
85- if isinstance (wanted_reference_keys , dict ): # when wanted_reference_keys is dict() type
107+ if isinstance (
108+ wanted_reference_keys ,
109+ dict ,
110+ ): # when wanted_reference_keys is dict() type
86111 list_of_reference_keys = list (wanted_reference_keys .keys ())
87112 elif any (
88113 isinstance (element , list ) for element in wanted_reference_keys
89114 ): # when wanted_reference_keys is a nested list
90115 list_of_reference_keys = flatten_list (wanted_reference_keys )[0 ]
91- elif isinstance (wanted_reference_keys , list ): # when wanted_reference_keys is a list
116+ elif isinstance (
117+ wanted_reference_keys ,
118+ list ,
119+ ): # when wanted_reference_keys is a list
92120 list_of_reference_keys = wanted_reference_keys
93121 else :
94- raise ValueError ("Reference Key normalization failure. Please verify data type returned." )
122+ raise ValueError (
123+ "Reference Key normalization failure. Please verify data type returned." ,
124+ )
95125
96- normalized = keys_values_zipper (list_of_reference_keys , paired_key_value )
126+ normalized = keys_values_zipper (
127+ list_of_reference_keys ,
128+ paired_key_value ,
129+ )
97130 # Data between pre and post may come in different order, so it needs to be sorted.
98131 return sorted (normalized , key = lambda arg : list (arg .keys ()))
99132
0 commit comments