Merge pull request #58 from PickwickSoft/feature/#57/add-csv-loader

Feature/#57/add csv loader

Author: garlontas

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "streams.py"
-version = "0.3.3"
+version = "1.0.0"
 authors = ["Stefan Garlonta <stefan@pickwicksoft.org>"]
 description = "A stream library for Python inspired by Java Stream API"
 keywords = ["streams", "parallel", "data"]

pystreamapi/__init__.py

@@ -1,5 +1,5 @@
 from pystreamapi.__stream import Stream
 from pystreamapi._streams.error.__levels import ErrorLevel
 
-__version__ = "0.3.3"
+__version__ = "1.0.0"
 __all__ = ["Stream", "ErrorLevel"]

pystreamapi/loaders/__csv_loader.py

@@ -0,0 +1,48 @@
+import contextlib
+import os
+from collections import namedtuple
+from csv import reader
+
+
+def csv(file_path: str, delimiter=',', encoding="utf-8") -> list:
+    """
+    Loads a CSV file and converts it into a list of namedtuples.
+
+    Returns:
+        list: A list of namedtuples, where each namedtuple represents a row in the CSV.
+        :param encoding: The encoding of the CSV file.
+        :param file_path: The path to the CSV file.
+        :param delimiter: The delimiter used in the CSV file.
+    """
+    file_path = __validate_path(file_path)
+    # skipcq: PTC-W6004
+    with open(file_path, 'r', newline='', encoding=encoding) as csvfile:
+        csvreader = reader(csvfile, delimiter=delimiter)
+
+        # Create a namedtuple type, casting the header values to int or float if possible
+        Row = namedtuple('Row', list(next(csvreader, [])))
+
+        # Process the data, casting values to int or float if possible
+        data = [Row(*[__try_cast(value) for value in row]) for row in csvreader]
+
+    return data
+
+
+def __validate_path(file_path: str):
+    """Validate a path string to prevent path traversal attacks"""
+    if not os.path.isabs(file_path):
+        raise ValueError("The file_path must be an absolute path.")
+
+    if not os.path.exists(file_path):
+        raise FileNotFoundError("The specified file does not exist.")
+
+    return file_path
+
+
+def __try_cast(value):
+    """Try to cast value to primary data types from python (int, float, bool)"""
+    for cast in (int, float):
+        with contextlib.suppress(ValueError):
+            return cast(value)
+    # Try to cast to bool
+    return value.lower() == 'true' if value.lower() in ('true', 'false') else value

pystreamapi/loaders/__init__.py

@@ -0,0 +1,5 @@
+from pystreamapi.loaders.__csv_loader import csv
+
+__all__ = [
+    'csv'
+]

tests/assets/data.csv

@@ -0,0 +1,3 @@
+attr1,attr2
+1,2.0
+a,b

tests/assets/data2.csv

@@ -0,0 +1,2 @@
+attr1;attr2
+1;2

tests/assets/empty.csv

@@ -0,0 +1,37 @@
+import os
+from unittest import TestCase
+from pystreamapi.loaders import csv
+
+class TestLoaders(TestCase):
+
+    def setUp(self) -> None:
+        cwd = os.path.dirname(os.path.realpath(__file__))
+        self.path = os.path.join(cwd, 'assets')
+
+    def test_csv_loader(self):
+        data = csv(f'{self.path}/data.csv')
+        self.assertEqual(len(data), 2)
+        self.assertEqual(data[0].attr1, 1)
+        self.assertIsInstance(data[0].attr1, int)
+        self.assertEqual(data[0].attr2, 2.0)
+        self.assertIsInstance(data[0].attr2, float)
+        self.assertEqual(data[1].attr1, 'a')
+        self.assertIsInstance(data[1].attr1, str)
+
+    def test_csv_loader_with_custom_delimiter(self):
+        data = csv(f'{self.path}/data2.csv', delimiter=';')
+        self.assertEqual(len(data), 1)
+        self.assertEqual(data[0].attr1, 1)
+        self.assertIsInstance(data[0].attr1, int)
+
+    def test_csv_loader_with_empty_file(self):
+        data = csv(f'{self.path}/empty.csv')
+        self.assertEqual(len(data), 0)
+
+    def test_csv_loader_with_invalid_path(self):
+        with self.assertRaises(FileNotFoundError):
+            csv(f'{self.path}/invalid.csv')
+
+    def test_csv_loader_with_non_absolute_path(self):
+        with self.assertRaises(ValueError):
+            csv('invalid.csv')