from typing import Any, List, Tuple, Union import pandas as pd from django.db.models import Choices, Model from ..alimentation_decorators import get_data_logger logger = get_data_logger(__name__) def batch_iterator(iterable: Union[List, Tuple, pd.DataFrame], batch_size: int) -> Union[List, Tuple, pd.DataFrame]: """ (je pense que le nom _batch_generator porterait à confusion) Générateur qui morcelle un itérable en lots de taille donnée. :param iterable: itérable (supported un data frame) :type iterable: Union[List,Tuple,pd.DataFrame] :param batch_size: taille de lot :type batch_size: int :return: lot (valide, utilisateur, nouveau) :rtype: Tuple[bool, Optional[UserType], bool] """ length = len(iterable) for idx in range(0, length, batch_size): if isinstance(iterable, pd.DataFrame): yield iterable.iloc[idx:min(idx + batch_size, length)] else: yield iterable[idx:min(idx + batch_size, length)] def is_same_value(val1: Any, val2: Any) -> bool: """ Indique si deux valeurs sont équivalentes. Le cas le plus courant est une instance de Choices comparée à sa valeur. """ _1 = val1 if isinstance(val1, Choices) else val1 _2 = val2 if isinstance(val2, Choices) else val2 return _1 == _2 def is_same_model(fields: Union[List[str], Tuple[str]], m1: Model, m2: Model) -> bool: """ Indique si deux modèles contiennent les mêmes données. Ce test permet de limiter les MAJ. """ def _get_value(m: Model, field: str): """ Récupère la valeur d'un champ """ attr = getattr(m, field) return attr.value if isinstance(attr, Choices) else attr return not all(is_same_value(getattr(m1, f), getattr(m2, f)) for f in fields) class InsertionCounters: """ conteneur de compteurs pour faciliter l'incrémentation dans des fonctions imbriquées """ ignored = 0 errors = 0