| 133 | self.validation_results: Dict = {} |
| 134 | |
| 135 | def extract( |
| 136 | self, |
| 137 | extractors: Dict[Text, Text], |
| 138 | variables_mapping: VariablesMapping = None, |
| 139 | ) -> Dict[Text, Any]: |
| 140 | if not extractors: |
| 141 | return {} |
| 142 | |
| 143 | extract_mapping = {} |
| 144 | for key, field in extractors.items(): |
| 145 | if "$" in field: |
| 146 | # field contains variable or function |
| 147 | field = self.parser.parse_data(field, variables_mapping) |
| 148 | field_value = self._search_jmespath(field) |
| 149 | extract_mapping[key] = field_value |
| 150 | |
| 151 | logger.info(f"extract mapping: {extract_mapping}") |
| 152 | return extract_mapping |
| 153 | |
| 154 | def _search_jmespath(self, expr: Text) -> Any: |
| 155 | try: |