def validate_partition_details()

in src/generate_query.py [0:0]

• 46 lines of code
• 13 McCabe index (conditional complexity)

def validate_partition_details(partition_details, load_frequency):
    if load_frequency == 'RUNTIME':
        e_msg = ('`partition_details` property should NOT be specified '
                 'for runtime views, but it IS specified.')
        return e_msg

    partition_column = partition_details.get('column')
    if not partition_column:
        e_msg = ('Partition `column` property missing from '
                 '`partition_details` property.')
        return e_msg

    partition_type = partition_details.get('partition_type')
    if not partition_type:
        e_msg = ('`partition_type` property missing from '
                 '`partition_details` property.')
        return e_msg

    if partition_type not in _PARTITION_TYPES:
        e_msg = ('`partition_type` has to be one of the following:'
                 f'{_PARTITION_TYPES}.\n'
                 f'Specified `partition_type` is "{partition_type}".')
        return e_msg

    if partition_type == 'time':
        time_partition_grain = partition_details.get('time_grain')
        if not time_partition_grain:
            e_msg = ('`time_grain` property missing for '
                     '`time` based partition.')
            return e_msg

        if time_partition_grain not in _TIME_PARTITION_GRAIN_LIST:
            e_msg = ('`time_grain` property has to be one of the following:'
                     f'{_TIME_PARTITION_GRAIN_LIST}.\n'
                     f'Specified `time_grain` is "{time_partition_grain}".')
            return e_msg

    if partition_type == 'integer_range':
        integer_range_bucket = partition_details.get('integer_range_bucket')
        if not integer_range_bucket:
            e_msg = ('`integer_range_bucket` property missing for '
                     '`integer_range` based partition.')
            return e_msg

        bucket_start = integer_range_bucket.get('start')
        bucket_end = integer_range_bucket.get('end')
        bucket_interval = integer_range_bucket.get('interval')

        if (bucket_start is None or bucket_end is None or
                bucket_interval is None):
            e_msg = ('Error: `start`, `end` or `interval` property missing for '
                     'the `integer_range_bucket` property.')
            return e_msg

    return None