in fbpcs/infra/cloud_bridge/data_validation/validation_utility/validation.py [0:0]
def validate_line(line: Dict[str, str], validation_state: ValidationState) -> None:
missing_required_field = False
missing_all_required_fields = len(validation_state.one_or_more_required_fields) > 0
pattern_validation_failed = False
current_line = validation_state.total_rows + HEADER_ROW_OFFSET
for field in validation_state.all_required_fields:
if field not in line or value_empty(line[field]):
missing_required_field = True
append_line_number_to_field(
field, validation_state.lines_missing_required_field, current_line
)
else:
field_is_valid = field_value_is_valid(
line[field], validation_state.format_validation_for_field[field]
)
if not field_is_valid:
pattern_validation_failed = True
append_line_number_to_field(
field, validation_state.lines_incorrect_field_format, current_line
)
for field in validation_state.one_or_more_required_fields:
if field in line and not value_empty(line[field]):
missing_all_required_fields = False
field_is_valid = field_value_is_valid(
line[field], validation_state.format_validation_for_field[field]
)
if not field_is_valid:
pattern_validation_failed = True
append_line_number_to_field(
field, validation_state.lines_incorrect_field_format, current_line
)
if (
missing_all_required_fields
and len(validation_state.lines_missing_all_required_fields) <= MAX_ERROR_LINES
):
validation_state.lines_missing_all_required_fields.append(current_line)
if (
missing_required_field
or missing_all_required_fields
or pattern_validation_failed
):
validation_state.error_rows += 1
else:
validation_state.valid_rows += 1