csv interface minor update to use the csv exceptions for errors
This commit is contained in:
@@ -7,6 +7,9 @@ from typing import Any, Sequence
|
||||
from pathlib import Path
|
||||
from collections import Counter
|
||||
import csv
|
||||
from corelibs.exceptions.csv_exceptions import (
|
||||
NoCsvReader, CompulsoryCsvHeaderCheckFailed, CsvHeaderDataMissing
|
||||
)
|
||||
|
||||
DELIMITER = ","
|
||||
QUOTECHAR = '"'
|
||||
@@ -50,17 +53,19 @@ class CsvWriter:
|
||||
csv.DictWriter[str] | None: _description_
|
||||
"""
|
||||
# if header order is set, make sure all header value fields exist
|
||||
if not self.header:
|
||||
raise CsvHeaderDataMissing("No header data available to write CSV file")
|
||||
header_values = self.header
|
||||
if header_order is not None:
|
||||
if Counter(header_values) != Counter(header_order):
|
||||
raise ValueError(
|
||||
raise CompulsoryCsvHeaderCheckFailed(
|
||||
"header order does not match header values: "
|
||||
f"{', '.join(header_values)} != {', '.join(header_order)}"
|
||||
)
|
||||
header_values = header_order
|
||||
# no duplicates
|
||||
if len(header_values) != len(set(header_values)):
|
||||
raise ValueError(f"Header must have unique values only: {', '.join(header_values)}")
|
||||
raise CompulsoryCsvHeaderCheckFailed(f"Header must have unique values only: {', '.join(header_values)}")
|
||||
try:
|
||||
fp = open(
|
||||
self.__file_name,
|
||||
@@ -76,7 +81,7 @@ class CsvWriter:
|
||||
csv_file_writer.writeheader()
|
||||
return csv_file_writer
|
||||
except OSError as err:
|
||||
raise ValueError(f"Could not open CSV file for writing: {err}") from err
|
||||
raise NoCsvReader(f"Could not open CSV file for writing: {err}") from err
|
||||
|
||||
def write_csv(self, line: dict[str, str]) -> None:
|
||||
"""
|
||||
@@ -135,14 +140,16 @@ class CsvReader:
|
||||
quoting=self.__quoting,
|
||||
)
|
||||
self.header = csv_file_reader.fieldnames
|
||||
if not self.header:
|
||||
raise CsvHeaderDataMissing("No header data available in CSV file")
|
||||
if self.__header_check is not None:
|
||||
header_diff = set(self.__header_check).difference(set(self.header or []))
|
||||
if header_diff:
|
||||
raise ValueError(
|
||||
raise CompulsoryCsvHeaderCheckFailed(
|
||||
f"CSV header does not match expected header: {', '.join(header_diff)} missing"
|
||||
)
|
||||
return csv_file_reader
|
||||
except OSError as err:
|
||||
raise ValueError(f"Could not open CSV file for reading: {err}") from err
|
||||
raise NoCsvReader(f"Could not open CSV file for reading: {err}") from err
|
||||
|
||||
# __END__
|
||||
|
||||
Reference in New Issue
Block a user