def _generate_ingestion_message()

in collection_manager/collection_manager/services/CollectionProcessor.py [0:0]

• 34 lines of code
• 10 McCabe index (conditional complexity)

    def _generate_ingestion_message(granule_path: str, collection: Collection) -> str:

        config_dict = {
            'granule': {
                'resource': granule_path
            },
            'slicer': {
                'name': 'sliceFileByStepSize',
                'dimension_step_sizes': dict(collection.slices)
            },
            'processors': CollectionProcessor._get_default_processors(collection)
        }

        group_vars = []

        for name, value in collection.dimension_names:
            if name != 'variable':
                continue
            else:
                value = json.loads(value)

                if isinstance(value, str):
                    value = [value]

                for v in value:
                    parts = v.split('/')

                    if len(parts) > 1:
                        group_vars.append(v)

        if collection.preprocess is not None:
            config_dict['preprocess'] = json.loads(collection.preprocess)

        if collection.processors is not None:
            config_dict['processors'].extend(json.loads(collection.processors))

        if collection.group is not None:
            config_dict['granule']['group'] = collection.group

        if len(group_vars) > 0:
            config_dict['granule']['grouped_vars'] = group_vars

        config_str = yaml.dump(config_dict)
        logger.debug(f"Templated dataset config:\n{config_str}")
        return config_str