65 lines
2.1 KiB
Python
65 lines
2.1 KiB
Python
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
|