时间戳

首页 > 解决方案 > NamedTuple 对象映射到 CSV 文件

python - NamedTuple 对象映射到 CSV 文件

问题描述

所以我有一个通过 NamedTuple() 验证 CSV 文件的对象...

# !/usr/bin/env python
# -*- coding: utf-8 -*-
import csv
import typing as T
import logging
from csv import DictReader
from collections import OrderedDict


class Validate(T.NamedTuple):

    field1: str
    field2: str
    field3: str


    @classmethod
    def from_row(cls, row: dict):
        return cls(**{key: type_(row[key]) for key, type_ in cls._field_types.items()})

    @staticmethod
    def validate_csv(reader: DictReader) -> bool:
        for index, row in enumerate(reader):
            try:
                PCW.from_row(row)
            except Exception as ERROR_MESSAGE:
                logging.error(f'{ERROR_MESSAGE} -- row_number: {index + 1}')
                return False
        return True

def execute(csv_file):
    input_file = csv.DictReader(open(csv_file))
    return Validate.validate_csv(input_file)


execute("../test_csv.csv")

对不起，如果这是一种设计模式，我对 OOP 编程还很陌生。

如果我们有以下形式的 CSV 文件，这将起作用：

field1,field2,field3

如果所有 3 个字段都属于该类型，那么它可以工作 - 如果任何一个属于不同类型，那么它将记录。这是预期的行为。

但是，我想做的是有某种方法一次只运行一行，每次调用它都会通过验证运行下一行。

IE：

def csv_file(filepath: str):
    with open(filepath) as file:
        reader = csv.DictReader(file, delimiter=',')
        for r in reader:
            yield r


for i in Validate.csv_file("../test_csv.csv"):
    Validate.validate_csv(i)

但是，这似乎不起作用 - 有谁知道我怎样才能让它工作？

标签： pythonpython-3.xcsvobject

解决方案

看起来我需要做的就是将整个阅读器传递给该from_row方法，这将获取每一行并针对 namedtuple 进行验证:)


(...)

 @classmethod
    def from_row(cls, row: dict):
        return cls(**{key: type_(row[key]) for key, type_ in cls._field_types.items()})


    @staticmethod
    def validate_csv(reader: dict) -> bool:
        try:
            PCW.from_row(reader)
        except Exception as e:
            logging.error(f"{e}")
            return False
        return True


def csv_file(filepath: str):
    with open(filepath) as file:
        reader = csv.DictReader(file, delimiter=',')
        for r in reader:
            yield r


for i in csv_file("../test_csv.csv"):
    PCW.validate_csv(i)

推荐阅读