Исходный код EJIO.utils.dataset.implementations.csv_storage

## /utils/dataset/implementations/csv_storage.py

import csv
from os.path import exists
from typing import Any
from ..dataset_storage import DatasetStorage
from ..dataset_registry import register_dataset
from ..err import DatasetError

__all__ = [
    'CSVStorage',
]




@register_dataset('csv')
class CSVStorage(DatasetStorage):
    """
    Реализация табличного хранилища для формата CSV.
    Поддерживает кастомные кодировки, разделители (delimiter) и диалекты.
    """
    __slots__ = ()



    @staticmethod
    def save[T](data: list[dict[str, T]], filename: str, **kwargs: Any) -> None:
        """
        Сохранить список словарей в CSV-файл.

        :param data: Список словарей для сериализации.
        :param filename: Путь к файлу для сохранения.
        :param kwargs: Дополнительные ключевые аргументы, для получения необязательных параметров.
        :raises DatasetError: При ошибке сохранения датасета.
        """
        if not data:
            try:
                with open(filename, 'w', encoding='utf-8') as f:
                    pass
                return
            except (IOError, OSError) as e:
                raise DatasetError(f"Failed to write empty CSV-file '{filename}': {e}") from e

        encoding = kwargs.pop('encoding', 'utf-8')
        delimiter = kwargs.pop('delimiter', ',')

        try:
            # Извлекаем заголовки колонок из ключей первого словаря
            fieldnames = list(data[0].keys())

            with open(filename, 'w', newline='', encoding=encoding) as f:
                writer = csv.DictWriter(f, fieldnames=fieldnames, delimiter=delimiter, **kwargs)
                writer.writeheader()
                writer.writerows(data)
        except (IOError, OSError, csv.Error) as e:
            raise DatasetError(f"Failed to write data set in CSV-file '{filename}': {e}") from e




    @staticmethod
    def load[T](filename: str, **kwargs: Any) -> list[dict[str, T]]:
        """
        Загрузить CSV-файл в виде списка словарей.

        :param filename: Путь к файлу для загрузки.
        :param kwargs: Дополнительные ключевые аргументы, для получения необязательных параметров.
        :return: Список словарей.
        :raises DatasetError: При ошибке загрузки датасета.
        """
        if not exists(filename):
            raise FileNotFoundError(f"Table file '{filename}' not found.")

        encoding = kwargs.pop('encoding', 'utf-8')
        delimiter = kwargs.pop('delimiter', ',')

        try:
            with open(filename, 'r', newline='', encoding=encoding) as f:
                reader = csv.DictReader(f, delimiter=delimiter, **kwargs)
                # Возвращаем строго типизированный список словарей
                return [dict(row) for row in reader]  # type: ignore[misc, return-value]
        except (IOError, OSError, csv.Error) as e:
            raise DatasetError(f"Failed to load data set from CSV-file '{filename}': {e}") from e