Work being done on #11

classes/importer/csv.py

@@ -0,0 +1,64 @@
+from codecs import decode
+from csv import DictReader
+from datetime import datetime
+from typing import Any, Dict, List, Union
+
+from bson import ObjectId
+
+from classes.importer.abstract import Importer
+from modules.database import col_entries
+
+
+class ImporterCSV(Importer):
+    """
+    The ImporterCSV class represents the object with
+    functionality to import/export garbage collection
+    records and convert them to other object types
+    from CSV files.
+    """
+
+    def __init__(self):
+        super(Importer, self).__init__()
+
+    async def import_data(self, data: bytes) -> List[ObjectId]:
+        entries: List[Dict[str, Any]] = list(
+            DictReader(decode(data).split("\n"), delimiter=";")
+        )
+
+        for entry in entries:
+            entry["locations"] = (
+                [int(entry["locations"])]
+                if "," not in entry["locations"]
+                else [int(id) for id in entry["locations"].split(",")]
+            )
+            entry["garbage_type"] = int(entry["garbage_type"])
+
+            for key in ("locations", "garbage_type", "date"):
+                if (
+                    key not in entry
+                    or (key == "garbage_type" and not isinstance(entry[key], int))
+                    or (key == "locations" and not isinstance(entry[key], list))
+                ):
+                    raise ValueError
+
+                if key == "date":
+                    try:
+                        datetime.fromisoformat(str(entry[key]))
+                    except (ValueError, TypeError) as exc:
+                        raise ValueError from exc
+
+        entries_clean: List[Dict[str, Union[str, int, datetime]]] = [
+            {
+                "locations": entry["locations"],
+                "garbage_type": entry["garbage_type"],
+                "date": datetime.fromisoformat(str(entry["date"])),
+            }
+            for entry in entries
+        ]
+
+        inserted = await col_entries.insert_many(entries_clean)
+
+        return [] if inserted is None else inserted.inserted_ids
+
+    async def export_data(self, data: Any) -> Any:
+        return None