TelegramBot/classes/importer/csv.py

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_api 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