def stage_b_transform()

in CommonLayerCode/datalake-library/python/datalake_library/transforms/transform_handler.py [0:0]

• 12 lines of code
• 7 McCabe index (conditional complexity)

    def stage_b_transform(self, bucket, keys, team, dataset): 
        """Applies StageB Transformation to Objects 
         
        Arguments: 
            bucket {string} -- Origin S3 Bucket 
            keys {string} -- Keys to transform
            team {string} -- Team owning the transformation  
            dataset {string} -- Dataset targeted by transformation
        Returns: 
            {dict} -- Dictionary of Bucket, Keys transformed, Path to Keys Processed and Job Details 
        """ 
        transform_info = self.get_transform_info('{}-{}'.format(team, dataset)) 
        module = import_module('datalake_library.transforms.stage_b_transforms.{}'.format(transform_info['stage_b_transform'])) 
        Transform = getattr(module, 'CustomTransform') 
        try: 
            response = Transform().transform_object(bucket, keys, team, dataset) 
        except Exception as e:
            raise e 
         
        if ((len(response) == 0) or (not isinstance(response, dict)) or ('processedKeysPath' not in response)
         or ('jobDetails' not in response) or ('jobStatus' not in response['jobDetails'])):
            raise ValueError("Invalid dictionary - Aborting") 
         
        return response