system-pri/backend/app/coordinator/utils.py

from typing import Generator, Any
from random import randint

import pandas as pd

from .exceptions import InvalidNameOrTypeHeaderException
from ..students.models import Student


def check_columns(df: pd.DataFrame) -> bool:
    headers = set(df.keys().values)
    columns = ['first_name', 'last_name', 'index', 'mode']
    if len(headers - set(columns)) != 0:
        return False

    flag = True
    col_types = ['object', 'object', 'int', 'int']

    for name, col_type in zip(columns, col_types):
        if not str(df.dtypes[name]).startswith(col_type):
            flag = False
            break

    return flag


def parse_csv(file, mode) -> Generator[Student, Any, None]:
    df = pd.read_csv(file)

    if not check_columns(df):
        raise InvalidNameOrTypeHeaderException

    students = (Student(**dict(item.items(), mode=mode))
                for _, item in df.iterrows())
    return students