def extract_graphql_data_items()

in connectors/sources/graphql.py [0:0]

• 23 lines of code
• 9 McCabe index (conditional complexity)

    def extract_graphql_data_items(self, data):
        """Returns sub objects from the response based on graphql_object_to_id_map

        Args:
            data (dict): data to extract

        Yields:
            dictionary/list: Documents from the response
        """
        for keys, field_id in self.graphql_object_to_id_map.items():
            current_level_data = data
            key_list = keys.split(".")

            for key in key_list:
                if isinstance(current_level_data, dict):
                    if key not in current_level_data.keys():
                        msg = (
                            f"{key} not found in the response. Please check the query."
                        )
                        raise ConfigurableFieldValueError(msg)
                    current_level_data = current_level_data.get(key, {})
                else:
                    msg = "Found list while processing GraphQL Objects List. Please check the query and provide path upto dictionary."
                    raise ConfigurableFieldValueError(msg)

            if isinstance(current_level_data, dict):
                if current_level_data:
                    current_level_data["_id"] = current_level_data.get(field_id)
                    yield current_level_data

            if isinstance(current_level_data, list):
                for doc in current_level_data:
                    doc["_id"] = doc.get(field_id)
                    yield doc