def validate_partition_columns()

in common/py_libs/bq_materializer.py [0:0]

• 24 lines of code
• 8 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:
        raise ValueError(
            f"Column '{column}' specified for partitioning does not exist in "
            "the table.")

    # 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):
        raise ValueError(
            "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.")

    if (partition_type == "integer_range" and
            partition_column_type != "INTEGER"):
        raise ValueError(
            "Error: For 'partition_type' = 'integer_range', partitioning "
            f"column has to be of INTEGER data type. But column '{column}' is "
            f"of '{partition_column_type}'.")