def validate_partition_columns()

in src/generate_query.py [0:0]

• 24 lines of code
• 4 McCabe index (conditional complexity)

def validate_partition_columns(partition_details, target_schema):
    """Checks schema to make sure columns are appropriate for partitioning."""

    column = partition_details['column']

    partition_column_details = [
        field for field in target_schema if field.name == column
    ]
    if not partition_column_details:
        e_msg = (f'Column "{column}" specified for partitioning does not '
                 'exist in the table.')
        raise cortex_exc.CriticalError(e_msg) from None

    # Since there will be only value in the list (a column exists only once
    # in a table), let's just use the first value from the list.
    partition_column_type = partition_column_details[0].field_type

    partition_type = partition_details['partition_type']

    if (partition_type == 'time' and
            partition_column_type not in _TIME_PARTITION_DATA_TYPES):
        e_msg = ('For `partition_type` = "time", partitioning column has to be '
                 'one of the following data types:'
                 f'{_TIME_PARTITION_DATA_TYPES}.\n'
                 f'But column "{column}" is of "{partition_column_type}" type.')
        raise cortex_exc.TypeCError(e_msg) from None

    if (partition_type == 'integer_range' and
            partition_column_type != 'INTEGER'):
        e_msg = ('Error: For `partition_type` = "integer_range", '
                 'partitioning column has to be of INTEGER data type.\n'
                 f'But column "{column}" is of {partition_column_type}.')
        raise cortex_exc.TypeCError(e_msg) from None