54 lines
1.2 KiB
Python
54 lines
1.2 KiB
Python
import csv
|
|
from enum import IntEnum
|
|
from lib.csv_loader import CSVLoader, CSVRow
|
|
from lib.csv_loader.csv_loader import BoolValuePair, CSVRowDefaultConfig
|
|
from lib.csv_loader.mapping_strategies import MappingStrategyByHeader
|
|
|
|
|
|
class ActionEnum(IntEnum):
|
|
INSERT = 0
|
|
UPDATE = 1
|
|
DELETE = 2
|
|
|
|
|
|
class AssetRow(CSVRow):
|
|
asset_action: ActionEnum
|
|
asset_id: str
|
|
brand_id: int
|
|
model_id: int
|
|
serial_number: str
|
|
is_active: bool
|
|
|
|
class Config(CSVRowDefaultConfig):
|
|
bool_value_pair = BoolValuePair(true="YES", false="NO")
|
|
|
|
|
|
def main():
|
|
with open("csv/assets.csv", "r") as f:
|
|
reader = csv.reader(f, delimiter=";")
|
|
|
|
mapping_strategy = MappingStrategyByHeader(model_cls=AssetRow)
|
|
|
|
csv_loader = CSVLoader[AssetRow](
|
|
reader=reader,
|
|
output_model_cls=AssetRow,
|
|
has_header=True,
|
|
aggregate_errors=True,
|
|
mapping_strategy=mapping_strategy,
|
|
)
|
|
|
|
result = csv_loader.read_rows()
|
|
|
|
"Results:"
|
|
for row in result.rows:
|
|
print(row)
|
|
|
|
if result.has_errors():
|
|
print("Errors:")
|
|
for error in result.errors:
|
|
print(f"Line: {error.line_number}: {error.original_error}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|