airbyte_cdk.sources.declarative.parsers.manifest_component_transformer
1# 2# Copyright (c) 2023 Airbyte, Inc., all rights reserved. 3# 4 5import copy 6import typing 7from typing import Any, Dict, Mapping, Optional 8 9PARAMETERS_STR = "$parameters" 10 11 12DEFAULT_MODEL_TYPES: Mapping[str, str] = { 13 # CompositeErrorHandler 14 "CompositeErrorHandler.error_handlers": "DefaultErrorHandler", 15 # CursorPagination 16 "CursorPagination.decoder": "JsonDecoder", 17 # DatetimeBasedCursor 18 "DatetimeBasedCursor.end_datetime": "MinMaxDatetime", 19 "DatetimeBasedCursor.end_time_option": "RequestOption", 20 "DatetimeBasedCursor.start_datetime": "MinMaxDatetime", 21 "DatetimeBasedCursor.start_time_option": "RequestOption", 22 # DeclarativeSource 23 "DeclarativeSource.check": "CheckStream", 24 "DeclarativeSource.spec": "Spec", 25 "DeclarativeSource.streams": "DeclarativeStream", 26 # DeclarativeStream 27 "DeclarativeStream.retriever": "SimpleRetriever", 28 "DeclarativeStream.schema_loader": "JsonFileSchemaLoader", 29 # DynamicDeclarativeStream 30 "DynamicDeclarativeStream.stream_template": "DeclarativeStream", 31 "DynamicDeclarativeStream.components_resolver": "ConfigComponentResolver", 32 # HttpComponentsResolver 33 "HttpComponentsResolver.retriever": "SimpleRetriever", 34 "HttpComponentsResolver.components_mapping": "ComponentMappingDefinition", 35 # ConfigComponentResolver 36 "ConfigComponentsResolver.stream_config": "StreamConfig", 37 "ConfigComponentsResolver.components_mapping": "ComponentMappingDefinition", 38 # DefaultErrorHandler 39 "DefaultErrorHandler.response_filters": "HttpResponseFilter", 40 # DefaultPaginator 41 "DefaultPaginator.decoder": "JsonDecoder", 42 "DefaultPaginator.page_size_option": "RequestOption", 43 # DpathExtractor 44 "DpathExtractor.decoder": "JsonDecoder", 45 "DpathExtractor.record_expander": "RecordExpander", 46 # HttpRequester 47 "HttpRequester.error_handler": "DefaultErrorHandler", 48 # ListPartitionRouter 49 "ListPartitionRouter.request_option": "RequestOption", 50 # ParentStreamConfig 51 "ParentStreamConfig.request_option": "RequestOption", 52 "ParentStreamConfig.stream": "DeclarativeStream", 53 # RecordExpander 54 "RecordExpander.parent_fields": "ParentFieldPath", 55 # RecordSelector 56 "RecordSelector.extractor": "DpathExtractor", 57 "RecordSelector.record_filter": "RecordFilter", 58 # SimpleRetriever 59 "SimpleRetriever.paginator": "NoPagination", 60 "SimpleRetriever.record_selector": "RecordSelector", 61 "SimpleRetriever.requester": "HttpRequester", 62 # SubstreamPartitionRouter 63 "SubstreamPartitionRouter.parent_stream_configs": "ParentStreamConfig", 64 # AddFields 65 "AddFields.fields": "AddedFieldDefinition", 66 # CustomPartitionRouter 67 "CustomPartitionRouter.parent_stream_configs": "ParentStreamConfig", 68 # DynamicSchemaLoader 69 "DynamicSchemaLoader.retriever": "SimpleRetriever", 70 # SchemaTypeIdentifier 71 "SchemaTypeIdentifier.types_map": "TypesMap", 72} 73 74# We retain a separate registry for custom components to automatically insert the type if it is missing. This is intended to 75# be a short term fix because once we have migrated, then type and class_name should be requirements for all custom components. 76CUSTOM_COMPONENTS_MAPPING: Mapping[str, str] = { 77 "CompositeErrorHandler.backoff_strategies": "CustomBackoffStrategy", 78 "DeclarativeStream.retriever": "CustomRetriever", 79 "DeclarativeStream.transformations": "CustomTransformation", 80 "DefaultErrorHandler.backoff_strategies": "CustomBackoffStrategy", 81 "DefaultPaginator.pagination_strategy": "CustomPaginationStrategy", 82 "HttpRequester.authenticator": "CustomAuthenticator", 83 "HttpRequester.error_handler": "CustomErrorHandler", 84 "RecordSelector.extractor": "CustomRecordExtractor", 85 "SimpleRetriever.partition_router": "CustomPartitionRouter", 86} 87 88 89class ManifestComponentTransformer: 90 def propagate_types_and_parameters( 91 self, 92 parent_field_identifier: str, 93 declarative_component: Mapping[str, Any], 94 parent_parameters: Mapping[str, Any], 95 use_parent_parameters: Optional[bool] = None, 96 ) -> Dict[str, Any]: 97 """ 98 Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the 99 default component type if it was not already present. The resulting transformed components are a deep copy of the input 100 components, not an in-place transformation. 101 102 :param declarative_component: The current component that is having type and parameters added 103 :param parent_field_identifier: The name of the field of the current component coming from the parent component 104 :param parent_parameters: The parameters set on parent components defined before the current component 105 :param use_parent_parameters: If set, parent parameters will be used as the source of truth when key names are the same 106 :return: A deep copy of the transformed component with types and parameters persisted to it 107 """ 108 propagated_component = dict(copy.deepcopy(declarative_component)) 109 if "type" not in propagated_component: 110 # If the component has class_name we assume that this is a reference to a custom component. This is a slight change to 111 # existing behavior because we originally allowed for either class or type to be specified. After the pydantic migration, 112 # class_name will only be a valid field on custom components and this change reflects that. I checked, and we currently 113 # have no low-code connectors that use class_name except for custom components. 114 if "class_name" in propagated_component: 115 found_type = CUSTOM_COMPONENTS_MAPPING.get(parent_field_identifier) 116 else: 117 found_type = DEFAULT_MODEL_TYPES.get(parent_field_identifier) 118 if found_type: 119 propagated_component["type"] = found_type 120 121 # Combines parameters defined at the current level with parameters from parent components. Parameters at the current 122 # level take precedence 123 current_parameters = dict(copy.deepcopy(parent_parameters)) 124 component_parameters = propagated_component.pop(PARAMETERS_STR, {}) 125 current_parameters = ( 126 {**component_parameters, **current_parameters} 127 if use_parent_parameters 128 else {**current_parameters, **component_parameters} 129 ) 130 131 # When there is no resolved type, we're not processing a component (likely a regular object) and don't need to propagate parameters 132 # When the type refers to a json schema, we're not processing a component as well. This check is currently imperfect as there could 133 # be json_schema are not objects but we believe this is not likely in our case because: 134 # * records are Mapping so objects hence SchemaLoader root should be an object 135 # * connection_specification is a Mapping 136 if self._is_json_schema_object(propagated_component): 137 return propagated_component 138 139 # For objects that don't have type check if their object fields have nested components which should have `$parameters` in it. 140 # For example, QueryProperties in requester.request_parameters, etc. 141 # Update propagated_component value with nested components with parent `$parameters` if needed and return propagated_component. 142 if "type" not in propagated_component: 143 if self._has_nested_components(propagated_component): 144 propagated_component = self._process_nested_components( 145 propagated_component, 146 parent_field_identifier, 147 current_parameters, 148 use_parent_parameters, 149 ) 150 return propagated_component 151 152 # Parameters should be applied to the current component fields with the existing field taking precedence over parameters if 153 # both exist 154 for parameter_key, parameter_value in current_parameters.items(): 155 propagated_component[parameter_key] = ( 156 propagated_component.get(parameter_key) or parameter_value 157 ) 158 159 for field_name, field_value in propagated_component.items(): 160 if isinstance(field_value, dict): 161 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 162 excluded_parameter = current_parameters.pop(field_name, None) 163 parent_type_field_identifier = f"{propagated_component.get('type')}.{field_name}" 164 propagated_component[field_name] = self.propagate_types_and_parameters( 165 parent_type_field_identifier, 166 field_value, 167 current_parameters, 168 use_parent_parameters=use_parent_parameters, 169 ) 170 if excluded_parameter: 171 current_parameters[field_name] = excluded_parameter 172 elif isinstance(field_value, typing.List): 173 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 174 excluded_parameter = current_parameters.pop(field_name, None) 175 for i, element in enumerate(field_value): 176 if isinstance(element, dict): 177 parent_type_field_identifier = ( 178 f"{propagated_component.get('type')}.{field_name}" 179 ) 180 field_value[i] = self.propagate_types_and_parameters( 181 parent_type_field_identifier, 182 element, 183 current_parameters, 184 use_parent_parameters=use_parent_parameters, 185 ) 186 if excluded_parameter: 187 current_parameters[field_name] = excluded_parameter 188 189 if current_parameters: 190 propagated_component[PARAMETERS_STR] = current_parameters 191 return propagated_component 192 193 @staticmethod 194 def _is_json_schema_object(propagated_component: Mapping[str, Any]) -> bool: 195 # Component types are always strings: a list is a JSON schema type union and a dict is a 196 # `properties` map that has a property named `type`. 197 component_type = propagated_component.get("type") 198 return component_type == "object" or isinstance(component_type, (list, dict)) 199 200 @staticmethod 201 def _has_nested_components(propagated_component: Dict[str, Any]) -> bool: 202 for k, v in propagated_component.items(): 203 if isinstance(v, dict) and v.get("type"): 204 return True 205 return False 206 207 def _process_nested_components( 208 self, 209 propagated_component: Dict[str, Any], 210 parent_field_identifier: str, 211 current_parameters: Mapping[str, Any], 212 use_parent_parameters: Optional[bool] = None, 213 ) -> Dict[str, Any]: 214 for field_name, field_value in propagated_component.items(): 215 if isinstance(field_value, dict) and field_value.get("type"): 216 nested_component_with_parameters = self.propagate_types_and_parameters( 217 parent_field_identifier, 218 field_value, 219 current_parameters, 220 use_parent_parameters=use_parent_parameters, 221 ) 222 propagated_component[field_name] = nested_component_with_parameters 223 224 return propagated_component
PARAMETERS_STR =
'$parameters'
DEFAULT_MODEL_TYPES: Mapping[str, str] =
{'CompositeErrorHandler.error_handlers': 'DefaultErrorHandler', 'CursorPagination.decoder': 'JsonDecoder', 'DatetimeBasedCursor.end_datetime': 'MinMaxDatetime', 'DatetimeBasedCursor.end_time_option': 'RequestOption', 'DatetimeBasedCursor.start_datetime': 'MinMaxDatetime', 'DatetimeBasedCursor.start_time_option': 'RequestOption', 'DeclarativeSource.check': 'CheckStream', 'DeclarativeSource.spec': 'Spec', 'DeclarativeSource.streams': 'DeclarativeStream', 'DeclarativeStream.retriever': 'SimpleRetriever', 'DeclarativeStream.schema_loader': 'JsonFileSchemaLoader', 'DynamicDeclarativeStream.stream_template': 'DeclarativeStream', 'DynamicDeclarativeStream.components_resolver': 'ConfigComponentResolver', 'HttpComponentsResolver.retriever': 'SimpleRetriever', 'HttpComponentsResolver.components_mapping': 'ComponentMappingDefinition', 'ConfigComponentsResolver.stream_config': 'StreamConfig', 'ConfigComponentsResolver.components_mapping': 'ComponentMappingDefinition', 'DefaultErrorHandler.response_filters': 'HttpResponseFilter', 'DefaultPaginator.decoder': 'JsonDecoder', 'DefaultPaginator.page_size_option': 'RequestOption', 'DpathExtractor.decoder': 'JsonDecoder', 'DpathExtractor.record_expander': 'RecordExpander', 'HttpRequester.error_handler': 'DefaultErrorHandler', 'ListPartitionRouter.request_option': 'RequestOption', 'ParentStreamConfig.request_option': 'RequestOption', 'ParentStreamConfig.stream': 'DeclarativeStream', 'RecordExpander.parent_fields': 'ParentFieldPath', 'RecordSelector.extractor': 'DpathExtractor', 'RecordSelector.record_filter': 'RecordFilter', 'SimpleRetriever.paginator': 'NoPagination', 'SimpleRetriever.record_selector': 'RecordSelector', 'SimpleRetriever.requester': 'HttpRequester', 'SubstreamPartitionRouter.parent_stream_configs': 'ParentStreamConfig', 'AddFields.fields': 'AddedFieldDefinition', 'CustomPartitionRouter.parent_stream_configs': 'ParentStreamConfig', 'DynamicSchemaLoader.retriever': 'SimpleRetriever', 'SchemaTypeIdentifier.types_map': 'TypesMap'}
CUSTOM_COMPONENTS_MAPPING: Mapping[str, str] =
{'CompositeErrorHandler.backoff_strategies': 'CustomBackoffStrategy', 'DeclarativeStream.retriever': 'CustomRetriever', 'DeclarativeStream.transformations': 'CustomTransformation', 'DefaultErrorHandler.backoff_strategies': 'CustomBackoffStrategy', 'DefaultPaginator.pagination_strategy': 'CustomPaginationStrategy', 'HttpRequester.authenticator': 'CustomAuthenticator', 'HttpRequester.error_handler': 'CustomErrorHandler', 'RecordSelector.extractor': 'CustomRecordExtractor', 'SimpleRetriever.partition_router': 'CustomPartitionRouter'}
class
ManifestComponentTransformer:
90class ManifestComponentTransformer: 91 def propagate_types_and_parameters( 92 self, 93 parent_field_identifier: str, 94 declarative_component: Mapping[str, Any], 95 parent_parameters: Mapping[str, Any], 96 use_parent_parameters: Optional[bool] = None, 97 ) -> Dict[str, Any]: 98 """ 99 Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the 100 default component type if it was not already present. The resulting transformed components are a deep copy of the input 101 components, not an in-place transformation. 102 103 :param declarative_component: The current component that is having type and parameters added 104 :param parent_field_identifier: The name of the field of the current component coming from the parent component 105 :param parent_parameters: The parameters set on parent components defined before the current component 106 :param use_parent_parameters: If set, parent parameters will be used as the source of truth when key names are the same 107 :return: A deep copy of the transformed component with types and parameters persisted to it 108 """ 109 propagated_component = dict(copy.deepcopy(declarative_component)) 110 if "type" not in propagated_component: 111 # If the component has class_name we assume that this is a reference to a custom component. This is a slight change to 112 # existing behavior because we originally allowed for either class or type to be specified. After the pydantic migration, 113 # class_name will only be a valid field on custom components and this change reflects that. I checked, and we currently 114 # have no low-code connectors that use class_name except for custom components. 115 if "class_name" in propagated_component: 116 found_type = CUSTOM_COMPONENTS_MAPPING.get(parent_field_identifier) 117 else: 118 found_type = DEFAULT_MODEL_TYPES.get(parent_field_identifier) 119 if found_type: 120 propagated_component["type"] = found_type 121 122 # Combines parameters defined at the current level with parameters from parent components. Parameters at the current 123 # level take precedence 124 current_parameters = dict(copy.deepcopy(parent_parameters)) 125 component_parameters = propagated_component.pop(PARAMETERS_STR, {}) 126 current_parameters = ( 127 {**component_parameters, **current_parameters} 128 if use_parent_parameters 129 else {**current_parameters, **component_parameters} 130 ) 131 132 # When there is no resolved type, we're not processing a component (likely a regular object) and don't need to propagate parameters 133 # When the type refers to a json schema, we're not processing a component as well. This check is currently imperfect as there could 134 # be json_schema are not objects but we believe this is not likely in our case because: 135 # * records are Mapping so objects hence SchemaLoader root should be an object 136 # * connection_specification is a Mapping 137 if self._is_json_schema_object(propagated_component): 138 return propagated_component 139 140 # For objects that don't have type check if their object fields have nested components which should have `$parameters` in it. 141 # For example, QueryProperties in requester.request_parameters, etc. 142 # Update propagated_component value with nested components with parent `$parameters` if needed and return propagated_component. 143 if "type" not in propagated_component: 144 if self._has_nested_components(propagated_component): 145 propagated_component = self._process_nested_components( 146 propagated_component, 147 parent_field_identifier, 148 current_parameters, 149 use_parent_parameters, 150 ) 151 return propagated_component 152 153 # Parameters should be applied to the current component fields with the existing field taking precedence over parameters if 154 # both exist 155 for parameter_key, parameter_value in current_parameters.items(): 156 propagated_component[parameter_key] = ( 157 propagated_component.get(parameter_key) or parameter_value 158 ) 159 160 for field_name, field_value in propagated_component.items(): 161 if isinstance(field_value, dict): 162 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 163 excluded_parameter = current_parameters.pop(field_name, None) 164 parent_type_field_identifier = f"{propagated_component.get('type')}.{field_name}" 165 propagated_component[field_name] = self.propagate_types_and_parameters( 166 parent_type_field_identifier, 167 field_value, 168 current_parameters, 169 use_parent_parameters=use_parent_parameters, 170 ) 171 if excluded_parameter: 172 current_parameters[field_name] = excluded_parameter 173 elif isinstance(field_value, typing.List): 174 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 175 excluded_parameter = current_parameters.pop(field_name, None) 176 for i, element in enumerate(field_value): 177 if isinstance(element, dict): 178 parent_type_field_identifier = ( 179 f"{propagated_component.get('type')}.{field_name}" 180 ) 181 field_value[i] = self.propagate_types_and_parameters( 182 parent_type_field_identifier, 183 element, 184 current_parameters, 185 use_parent_parameters=use_parent_parameters, 186 ) 187 if excluded_parameter: 188 current_parameters[field_name] = excluded_parameter 189 190 if current_parameters: 191 propagated_component[PARAMETERS_STR] = current_parameters 192 return propagated_component 193 194 @staticmethod 195 def _is_json_schema_object(propagated_component: Mapping[str, Any]) -> bool: 196 # Component types are always strings: a list is a JSON schema type union and a dict is a 197 # `properties` map that has a property named `type`. 198 component_type = propagated_component.get("type") 199 return component_type == "object" or isinstance(component_type, (list, dict)) 200 201 @staticmethod 202 def _has_nested_components(propagated_component: Dict[str, Any]) -> bool: 203 for k, v in propagated_component.items(): 204 if isinstance(v, dict) and v.get("type"): 205 return True 206 return False 207 208 def _process_nested_components( 209 self, 210 propagated_component: Dict[str, Any], 211 parent_field_identifier: str, 212 current_parameters: Mapping[str, Any], 213 use_parent_parameters: Optional[bool] = None, 214 ) -> Dict[str, Any]: 215 for field_name, field_value in propagated_component.items(): 216 if isinstance(field_value, dict) and field_value.get("type"): 217 nested_component_with_parameters = self.propagate_types_and_parameters( 218 parent_field_identifier, 219 field_value, 220 current_parameters, 221 use_parent_parameters=use_parent_parameters, 222 ) 223 propagated_component[field_name] = nested_component_with_parameters 224 225 return propagated_component
def
propagate_types_and_parameters( self, parent_field_identifier: str, declarative_component: Mapping[str, Any], parent_parameters: Mapping[str, Any], use_parent_parameters: Optional[bool] = None) -> Dict[str, Any]:
91 def propagate_types_and_parameters( 92 self, 93 parent_field_identifier: str, 94 declarative_component: Mapping[str, Any], 95 parent_parameters: Mapping[str, Any], 96 use_parent_parameters: Optional[bool] = None, 97 ) -> Dict[str, Any]: 98 """ 99 Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the 100 default component type if it was not already present. The resulting transformed components are a deep copy of the input 101 components, not an in-place transformation. 102 103 :param declarative_component: The current component that is having type and parameters added 104 :param parent_field_identifier: The name of the field of the current component coming from the parent component 105 :param parent_parameters: The parameters set on parent components defined before the current component 106 :param use_parent_parameters: If set, parent parameters will be used as the source of truth when key names are the same 107 :return: A deep copy of the transformed component with types and parameters persisted to it 108 """ 109 propagated_component = dict(copy.deepcopy(declarative_component)) 110 if "type" not in propagated_component: 111 # If the component has class_name we assume that this is a reference to a custom component. This is a slight change to 112 # existing behavior because we originally allowed for either class or type to be specified. After the pydantic migration, 113 # class_name will only be a valid field on custom components and this change reflects that. I checked, and we currently 114 # have no low-code connectors that use class_name except for custom components. 115 if "class_name" in propagated_component: 116 found_type = CUSTOM_COMPONENTS_MAPPING.get(parent_field_identifier) 117 else: 118 found_type = DEFAULT_MODEL_TYPES.get(parent_field_identifier) 119 if found_type: 120 propagated_component["type"] = found_type 121 122 # Combines parameters defined at the current level with parameters from parent components. Parameters at the current 123 # level take precedence 124 current_parameters = dict(copy.deepcopy(parent_parameters)) 125 component_parameters = propagated_component.pop(PARAMETERS_STR, {}) 126 current_parameters = ( 127 {**component_parameters, **current_parameters} 128 if use_parent_parameters 129 else {**current_parameters, **component_parameters} 130 ) 131 132 # When there is no resolved type, we're not processing a component (likely a regular object) and don't need to propagate parameters 133 # When the type refers to a json schema, we're not processing a component as well. This check is currently imperfect as there could 134 # be json_schema are not objects but we believe this is not likely in our case because: 135 # * records are Mapping so objects hence SchemaLoader root should be an object 136 # * connection_specification is a Mapping 137 if self._is_json_schema_object(propagated_component): 138 return propagated_component 139 140 # For objects that don't have type check if their object fields have nested components which should have `$parameters` in it. 141 # For example, QueryProperties in requester.request_parameters, etc. 142 # Update propagated_component value with nested components with parent `$parameters` if needed and return propagated_component. 143 if "type" not in propagated_component: 144 if self._has_nested_components(propagated_component): 145 propagated_component = self._process_nested_components( 146 propagated_component, 147 parent_field_identifier, 148 current_parameters, 149 use_parent_parameters, 150 ) 151 return propagated_component 152 153 # Parameters should be applied to the current component fields with the existing field taking precedence over parameters if 154 # both exist 155 for parameter_key, parameter_value in current_parameters.items(): 156 propagated_component[parameter_key] = ( 157 propagated_component.get(parameter_key) or parameter_value 158 ) 159 160 for field_name, field_value in propagated_component.items(): 161 if isinstance(field_value, dict): 162 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 163 excluded_parameter = current_parameters.pop(field_name, None) 164 parent_type_field_identifier = f"{propagated_component.get('type')}.{field_name}" 165 propagated_component[field_name] = self.propagate_types_and_parameters( 166 parent_type_field_identifier, 167 field_value, 168 current_parameters, 169 use_parent_parameters=use_parent_parameters, 170 ) 171 if excluded_parameter: 172 current_parameters[field_name] = excluded_parameter 173 elif isinstance(field_value, typing.List): 174 # We exclude propagating a parameter that matches the current field name because that would result in an infinite cycle 175 excluded_parameter = current_parameters.pop(field_name, None) 176 for i, element in enumerate(field_value): 177 if isinstance(element, dict): 178 parent_type_field_identifier = ( 179 f"{propagated_component.get('type')}.{field_name}" 180 ) 181 field_value[i] = self.propagate_types_and_parameters( 182 parent_type_field_identifier, 183 element, 184 current_parameters, 185 use_parent_parameters=use_parent_parameters, 186 ) 187 if excluded_parameter: 188 current_parameters[field_name] = excluded_parameter 189 190 if current_parameters: 191 propagated_component[PARAMETERS_STR] = current_parameters 192 return propagated_component
Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the default component type if it was not already present. The resulting transformed components are a deep copy of the input components, not an in-place transformation.
Parameters
- declarative_component: The current component that is having type and parameters added
- parent_field_identifier: The name of the field of the current component coming from the parent component
- parent_parameters: The parameters set on parent components defined before the current component
- use_parent_parameters: If set, parent parameters will be used as the source of truth when key names are the same
Returns
A deep copy of the transformed component with types and parameters persisted to it