使用 Pyvaru 进行批量数据 (CSV) 验证

Question

我正在寻找一个通用的验证器模块来帮助清理数据，重要的是，返回一个错误日志，说明数据被拒绝的原因。我主要使用 CSV 文件，每个文件平均有 40 列和大约 40,000 行。 CSV 文件将包含个人身份信息、联系信息以及他们在我们这里持有的帐户的详细信息

例如

First Name|Last Name|Other Name|Passport Number|Date of Birth|Phone Number|Email Address|Physical Address|Account Number|Invoice Number|Date Opened|Amount Due|Date Due|etc|etc

我需要验证data type、data length、options/choices、ranges、mandatory fields 等基本内容。还有条件验证，例如如果已提供 Amount Due 值，则还必须提供 Date Due。如果没有，我会报错。

Pyvaru 提供了一些基本的验证类。是否可以使用 pyvaru 实现基本验证和条件验证这两种场景？如果是，我将如何构建验证。我必须创建对象，例如标识符对象，然后是帐户对象让我使用 pyvaru？

Answer 1

Pyvaru 验证 python 对象（类实例和集合，如字典、列表等），因此从 CSV 开始，我将使用 DictReader 将每条记录转换为字典。 所以，给定一个像这样的 CSV：

policyID,statecode,county,eq_site_limit,hu_site_limit,fl_site_limit,fr_site_limit,tiv_2011,tiv_2012,eq_site_deductible,hu_site_deductible,fl_site_deductible,fr_site_deductible,point_latitude,point_longitude,line,construction,point_granularity
119736,FL,CLAY COUNTY,498960,498960,498960,498960,498960,792148.9,0,9979.2,0,0,30.102261,-81.711777,Residential,Masonry,1
448094,FL,CLAY COUNTY,1322376.3,1322376.3,1322376.3,1322376.3,1322376.3,1438163.57,0,0,0,0,30.063936,-81.707664,Residential,Masonry,3
206893,FL,CLAY COUNTY,190724.4,190724.4,190724.4,190724.4,190724.4,192476.78,0,0,0,0,30.089579,-81.700455,Residential,Wood,1
333743,FL,CLAY COUNTY,0,79520.76,0,0,79520.76,86854.48,0,0,0,0,30.063236,-81.707703,Residential,Wood,3
172534,FL,CLAY COUNTY,0,254281.5,0,254281.5,254281.5,246144.49,0,0,0,0,30.060614,-81.702675,Residential,Wood,1

验证码类似于：

import csv
from pyvaru import Validator
from pyvaru.rules import MaxLengthRule, MaxValueRule


class CsvRecordValidator(Validator):
    def get_rules(self) -> list:
        record: dict = self.data
        return [
            MaxLengthRule(apply_to=record.get('statecode'),
                          label='State Code',
                          max_length=2),
            MaxValueRule(apply_to=record.get('eq_site_limit'),
                         label='Site limit',
                         max_value=40000),
        ]


with open('sample.csv', 'r') as csv_file:
    reader = csv.DictReader(csv_file)
    row = 0
    for record in reader:
        row += 1
        validator = CsvRecordValidator(record)
        validation = validator.validate()
        if not validation.is_successful():
            print(f'Row {row} did not validate. Details: {validation.errors}')

上面的例子只是你可以做的一个愚蠢的例子，特别是它检查“statecode”列的最大长度为 2 并且“eq_site_limit”的最大值为 40k。 您可以通过继承抽象类ValidationRule 并实现apply() 方法来实现自己的规则：

class ContainsHelloRule(ValidationRule):
    def apply(self) -> bool:
        return 'hello' in self.apply_to

也可以使用按位运算符来否定规则...例如，通过使用前面的自定义规则检查字符串 mst 是否包含“hello”，您可以编写：

~ ContainsHelloRule(apply_to=a_string, label="A test string")

因此该规则仅在字符串不包含“hello”字符串时才有效。

也可以验证 CSV 记录，无需字典转换，通过使用带有验证正则表达式的 PatternRule 验证每一行...但当然您不会知道哪个列值无效。