src/csv_worker.py - master - alexkorolex/Test_Case_CSV

src/csv_worker.py
86 строк3 KB
Alex Korolev
Delete schema product
13 окт 2025, 21:30
13 окт 2025, 21:30f210176
import csv
from pathlib import PosixPath
from typing import Any


def get_data_from_files(files: list[PosixPath]) -> list[dict[str, Any]]:
    """Function to get data from csv

    Args:
        files (list[PosixPath]): list of Path subclass for non-Windows systems.

    Raises:
        FileNotFoundError: If file not found
        ValueError: if file isn't csv
        ValueError: if data in file with error

    Returns:
        list[dict[str, Any]]: list product schema.
    """
    data: list[dict[str, Any]] = []
    for file in files:
        if not file.exists():
            raise FileNotFoundError(f"File not found: {str(file)}. Check file path.")
        if file.suffix.lower() != ".csv":
            raise ValueError(f"File {str(file)} isn't CSV.")
        with open(
            file=str(file),
            mode="r",
            encoding="utf-8",
        ) as f:
            reader = csv.DictReader(f)
            for row in reader:
                if any(value is None or value == "" for value in row.values()):
                    raise ValueError(
                        f"Uncorrect data in {file}. Error on row: {row}. Please check file data."
                    )
                data.append(row)
    return data


def filter_data(
    data: list[dict[str, Any]], keys_data: list[str]
) -> list[dict[str, Any]]:
    """
    Filters data based on the specified keys.

    Args:
        data (List[ProductInputSchema]): list of objects with the original data
        keys_data (List[str]): list of keys to retain

    Returns:
        List[Dict[str, Any]]: list of dictionaries containing only the selected fields
    """
    sample_keys = data[0].keys()
    for key in keys_data:
        if key not in sample_keys:
            raise ValueError(f"Ключ '{key}' отсутствует в данных!")

    filtered_list = []
    for row in data:
        row_dict = row
        filtered_row = {key: row_dict[key] for key in keys_data}
        filtered_list.append(filtered_row)
    return filtered_list


def save_data_to_report(
    data_filter: list[dict[str, Any]],
    path_to_file_report: PosixPath,
    keys_data: list[str],
) -> bool:
    """
    Args:
        data_filter (list[dict[str, Any]]): list of dictionaries containing the filtered data to save
        path_to_file_report (PosixPath): path to the CSV file where the report will be saved
        keys_data (list[str]): list of keys/fields to include in the report

    Returns:
        bool: True.
    """
    path_to_file_report.parent.mkdir(parents=True, exist_ok=True)
    with open(path_to_file_report, mode="w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=keys_data)
        writer.writeheader()
        writer.writerows(data_filter)
    return True