# CSV file parser with some checks for header fields and duplicate rows

I have built a sample Parser class on top of the csv module for basic checks that we normally need as the first step of pre-processing. I would love a bit of feedback on the code and any areas that I can improve upon in the code to make it more efficient.

import csv

class CsvParser:
def __init__(self, file: str, compulsory_headers: list):
self.file = file
self.duplicate_rows = []
self.good_rows = []
self.log = []

return True if not self.diff_headers() else False

def is_bad_row(self, line_num: int, curr_row: tuple):
if not all(curr_row):
return True
return False

def _append_log(self, line_num: int):
self.log.append({"total_num_of_rows_processed": line_num,
"good_rows": self.good_rows,
"duplicate_rows": self.duplicate_rows,
})

@property
def row(self):
next(rows)
for _row in rows:
yield _row

@property
def col(self):
for row in rows:
return row

def _parse_rows(self):
_rows = set()
next(rows)
line_num = 1
_rows = set()
for row in rows:
line_num += 1
row = tuple(row)
if row not in _rows:
self.good_rows.append({"line_num": line_num, "row": list(row)})
else:
self.duplicate_rows.append({"line_num": line_num,
"row": list(row)
})
self._append_log(line_num)

self._parse_rows()
else:
self._parse_rows()
else:

csv_parser = CsvParser("hello.csv", compulsory_headers=["id", "student", "marks", ])
print(csv_parser.process_csv())
print(csv_parser.good_rows)
print(csv_parser.duplicate_rows)
print(csv_parser.log)

• Please see What to do when someone answers. I have rolled back Rev 4 → 2. Apr 21 at 8:54
• @200_success you've got answering, editing, and modding on the same Q&A. There should be an achievement/badge for this. :) Apr 25 at 17:15

## Interface

The interface for this class is awkward, and it's not clear how it's meant to be used.

• Is diff_headers() supposed to be part of the interface? How about passes_compulsory_header_check()? Maybe they should be considered private, to be used internally by process_csv()?
• The is_bad_row(line_num, curr_row) method definitely seems like it's supposed to be a private method for internal use: how else would a caller come up with a line number and a tuple of data to pass to it? The worst aspect of this method is that even though its name suggests that it just performs a test, it actually has a side-effect of maybe appending something to self.bad_rows.
• The _append_log() method appends to self.log, but considering that the "log" should contain a single item that is a summary of the outcome of parsing the entire file, why is self.log a list at all? For that matter, why should there be a "log" at all? As your sample usage shows, you can get that information from csv_parser.good_rows, csv_parser.bad_rows, etc. — all that's missing is the line count, and you could modify the class so that the line count could be queried likewise.
• It's highly unorthodox that row() is a property that is also a generator — especially since the name of the method is singular rather than plural.
• I don't know what the col() method is for, but it looks entirely wrong, and you never call it.
• The process_csv() method looks somewhat reasonable. But why is there a skip_header_check option? Couldn't the caller also effectively skip the header check by specifying no compulsory headers? It seems simpler and more Pythonic to offer just one way to accomplish a particular goal.

## Design suggestion

If I had to write this class, I'd design it so that it works pretty much as a drop-in replacement for the csv.DictReader class, such that iterating over the reader streams the "good" rows, but keeps track of the bad and duplicate rows that it encounters.

## Implementation

In several places, you write reader = csv.reader(open(self.file)), often followed by next(rows) to skip the header row. It would be better to design the class so that it opens the file just once, and makes one linear pass through its contents. Also, a csv.DictReader would be a better way to handle the header row.

What exactly constitutes a "bad" row? A row that has an empty field? Your criterion is all(curr_row), but you should be aware that if the row has fewer fields than expected, that will be considered acceptable.

In _parse_rows(), you initialized _rows = set() twice. To write a for loop that maintains a count while iterating, the Python idiom is to use enumerate().

• Thank you for the feedback. I will refactor the code with the above suggestions and do an update here with the updated code.
– SDRJ
Apr 21 at 7:17

Based on the feedback from 200_success, here is revised version of the code

import csv

class CsvParser:
def __init__(self, file: str, compulsory_headers: list):
self.file = file
self.duplicate_rows = []
self.good_rows = []
self.total_lines_processed = None
self.num_of_columns = None

return True if not self._diff_headers() else False

def _parse_rows(self):
_rows = set()
line_num = None
for line_num, row in enumerate(rows, start=1):
row = tuple(row.values())
if all(row) and self.num_of_columns == len(row):
if row not in _rows:
self.good_rows.append({"line_num": line_num, "row": list(row)})
else:
self.duplicate_rows.append({"line_num": line_num,
"row": list(row)
})
else:
self.total_lines_processed = line_num

def process_csv(self):
self._parse_rows()
else:
self._parse_rows()
else:

csv_parser = CsvParser("hello.csv", compulsory_headers=["id", "student", "marks", ])
csv_parser.process_csv()
print(csv_parser.good_rows)

• The if not self.compulsory_headers: … special case is a bad idea. Do you see how that codepath can cause a crash? If I recall correctly, DictReader already enforces that each row contains the expected number of fields. Apr 21 at 14:45