Validazione con pydantic e pandera

[xyffar]

import pandera as pa
import pandas as pd
from pydantic import BaseModel, field_validator, Field
from pandera.typing import Series, DataFrame
from typing import List, Tuple
from datetime import datetime
import re

# Schema per la Series: valori float positivi
class PositiveFloatSeries(pa.SeriesModel):
    class Config:
        dtype = pa.Float
        checks = pa.Check.ge(0)
        index = pa.Index(pa.DateTime)

# Schema per il DataFrame: indice di tipo data e una colonna di serie
class DatedDataFrame(pa.DataFrameModel):
    valore_di_interesse: pa.Series[float] = pa.Field(nullable=True)
    eq_line: pa.Series[Series[PositiveFloatSeries]]

    class Config:
        index = pa.Index(datetime, name="data_ora", coerce=True)
        # La colonna 'valore_di_interesse' non deve contenere valori negativi
        checks = pa.Check.ge(0, column='valore_di_interesse')

# Modello per la validazione di ogni singolo dizionario
class ExtraDataModel(BaseModel):
    index: int = Field(gt=0)
    t_0: int = Field(gt=0)
    t_1: int
    date_0: datetime
    date_1: datetime

    @field_validator('t_1')
    @classmethod
    def validate_t_1(cls, v, info):
        if v <= info.data['t_0']:
            raise ValueError('t_1 deve essere maggiore di t_0')
        return v

    @field_validator('date_1')
    @classmethod
    def validate_date_1(cls, v, info):
        if v <= info.data['date_0']:
            raise ValueError('date_1 deve essere maggiore di date_0')
        return v

# Modello Pydantic che integra tutti gli schemi e le regole
class InputModel(BaseModel):
    my_int: int = Field(gt=0)
    my_float: float = Field(gt=0, lt=1)
    my_str: str
    my_list: List[int] = Field(min_length=17, max_length=17)
    my_tuple: Tuple[str, ...]
    my_series: Series[PositiveFloatSeries]
    my_dataframe: DataFrame[DatedDataFrame]
    
    # Lista di dizionari con le regole di validazione
    extra_data: List[ExtraDataModel]

    # Validatore per la stringa
    @field_validator('my_str')
    @classmethod
    def validate_my_str(cls, v):
        if not v:
            raise ValueError('La stringa non può essere vuota')
        if not re.match(r'^[a-zA-Z0-9.]+$', v):
            raise ValueError('La stringa può contenere solo lettere, numeri e il punto')
        return v
    
    # Validatore per la lista
    @field_validator('my_list')
    @classmethod
    def validate_my_list(cls, v):
        for item in v:
            if not (5 <= item <= 27):
                raise ValueError('Tutti i valori nella lista devono essere compresi tra 5 e 27')
        return v

def process_data(inputs: InputModel):
    print("Dati ricevuti e validati con successo.")
    print("---------------------------------")
    print("Informazioni sugli input:")
    print(f"Intero: {inputs.my_int}")
    print(f"Stringa: {inputs.my_str}")
    print(f"Serie Pandas:\n{inputs.my_series}")
    print(f"DataFrame Pandas:\n{inputs.my_dataframe}")
    print("\nInformazioni aggiuntive (lista di dizionari):")
    for i, data in enumerate(inputs.extra_data):
        print(f"  Dizionario {i+1}:")
        print(f"    Index: {data.index}")
        print(f"    t_0: {data.t_0}")
        print(f"    t_1: {data.t_1}")
        print(f"    date_0: {data.date_0}")
        print(f"    date_1: {data.date_1}")

# Esempio 1: Dati validi
valid_series = pd.Series([1.5, 2.0, 5.8], name='serie_valori')
valid_series_list = [
    pd.Series([10.1, 10.2], name='sub_series_1'),
    pd.Series([20.1, 20.2], name='sub_series_2')
]
valid_df = pd.DataFrame(
    {'valore_di_interesse': [10.5, 20.8],
     'eq_line': valid_series_list},
    index=pd.to_datetime(['2025-09-01', '2025-09-02'])
)

valid_extra_data_list = [
    {"index": 1, "t_0": 5, "t_1": 10, "date_0": datetime(2025, 1, 1), "date_1": datetime(2025, 1, 10)},
    {"index": 2, "t_0": 10, "t_1": 20, "date_0": datetime(2025, 2, 1), "date_1": datetime(2025, 2, 20)}
]

valid_data = {
    "my_int": 1,
    "my_float": 0.5,
    "my_str": 'testo.prova.123',
    "my_list": [10, 15, 20, 25, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
    "my_tuple": ('a', 'b'),
    "my_series": valid_series,
    "my_dataframe": valid_df,
    "extra_data": valid_extra_data_list
}

try:
    print("--- Tentativo con dati validi ---")
    valid_inputs = InputModel(**valid_data)
    process_data(valid_inputs)
except Exception as e:
    print(f"\nErrore inatteso: {e}")

print("\n\n" + "="*40 + "\n\n")

# Esempio 2: Dati non validi (valore negativo nella serie)
invalid_series_1 = pd.Series([1.5, -2.0, 5.8], name='serie_valori')
invalid_data_1 = {**valid_data, 'my_series': invalid_series_1}

try:
    print("--- Tentativo con dati non validi (serie) ---")
    InputModel(**invalid_data_1)
except Exception as e:
    print(f"\nErrore di validazione previsto:\n{e}")

print("\n\n" + "="*40 + "\n\n")

# Esempio 3: Dati non validi (t_1 non è maggiore di t_0 in uno dei dizionari)
invalid_extra_data_list = [
    {"index": 1, "t_0": 5, "t_1": 10, "date_0": datetime(2025, 1, 1), "date_1": datetime(2025, 1, 10)},
    {"index": 2, "t_0": 20, "t_1": 10, "date_0": datetime(2025, 2, 1), "date_1": datetime(2025, 2, 20)}  # t_1 < t_0
]
invalid_data_2 = {**valid_data, 'extra_data': invalid_extra_data_list}

try:
    print("--- Tentativo con dati non validi (lista di dizionari) ---")
    InputModel(**invalid_data_2)
except Exception as e:
    print(f"\nErrore di validazione previsto:\n{e}")

[xyffar]

Se la serie di serie dà problemi, prova così

def create_nested_series_schema(inner_schema_class):
    """Factory per creare schemi con Series nested validate"""
    
    def validate_with_schema(series_obj):
        try:
            inner_schema_class.validate(series_obj)
            return True
        except:
            return False
    
    return pa.Check(validate_with_schema, element_wise=True)

class DatedDataFrameV5(pa.DataFrameModel):
    valore_di_interesse: Series[float] = pa.Field(ge=0, nullable=True)
    eq_line: Series[object] = pa.Field(
        dtype=object,
        checks=[create_nested_series_schema(PositiveFloatSeries)]
    )
    
    class Config:
        index = pa.Index(datetime, name="data_ora", coerce=True)

[xyffar]

import pandera as pa
import pandas as pd
from pydantic import BaseModel, field_validator, Field, ValidationError
from pandera.typing import Series, DataFrame
from typing import List, Tuple
from datetime import datetime
import re
import logging

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Schema per la Series: valori float positivi
class PositiveFloatSeries(pa.SeriesModel):
    class Config:
        dtype = pa.Float64
        checks = [pa.Check.ge(0)]
        index = pa.Index(pa.DateTime)

# Factory Function per creare validatori di Series nested
def create_nested_series_schema(inner_schema_class):
    """
    Factory per creare schemi con Series nested validate.
    
    Args:
        inner_schema_class: La classe schema Pandera da applicare a ogni Series
    
    Returns:
        pa.Check: Un check Pandera che valida ogni elemento come Series
    """
    def validate_with_schema(series_obj):
        try:
            if not isinstance(series_obj, pd.Series):
                return False
            inner_schema_class.validate(series_obj)
            return True
        except Exception:
            return False
    
    return pa.Check(
        validate_with_schema, 
        element_wise=True,
        error=f"Each element must be a valid {inner_schema_class.__name__}"
    )

# Schema per il DataFrame: indice di tipo data e colonne validate
class DatedDataFrame(pa.DataFrameModel):
    valore_di_interesse: Series[float] = pa.Field(ge=0, nullable=True)
    eq_line: Series[object] = pa.Field(
        dtype=object,
        checks=[create_nested_series_schema(PositiveFloatSeries)]
    )

    class Config:
        index = pa.Index(datetime, name="data_ora", coerce=True)

# Modello per la validazione di ogni singolo dizionario
class ExtraDataModel(BaseModel):
    index: int = Field(gt=0, description="Indice identificativo positivo")
    t_0: int = Field(gt=0, description="Tempo iniziale positivo")
    t_1: int = Field(description="Tempo finale")
    date_0: datetime = Field(description="Data iniziale")
    date_1: datetime = Field(description="Data finale")

    @field_validator('t_1')
    @classmethod
    def validate_t_1(cls, v, info):
        if 't_0' in info.data and v <= info.data['t_0']:
            raise ValueError('t_1 deve essere maggiore di t_0')
        return v

    @field_validator('date_1')
    @classmethod
    def validate_date_1(cls, v, info):
        if 'date_0' in info.data and v <= info.data['date_0']:
            raise ValueError('date_1 deve essere maggiore di date_0')
        return v

# Modello Pydantic che integra tutti gli schemi e le regole
class InputModel(BaseModel):
    """
    Modello principale per la validazione di input complessi.
    
    Attributes:
        my_int: Intero positivo per identificazione
        my_float: Valore percentuale tra 0 e 1 (esclusivo)
        my_str: Identificatore alfanumerico con punti
        my_list: Lista di esattamente 17 interi compresi tra 5 e 27
        my_tuple: Tupla di stringhe
        my_series: Serie temporale con valori positivi
        my_dataframe: DataFrame con dati temporizzati e Series nested
        extra_data: Metadati aggiuntivi con vincoli temporali
    """
    my_int: int = Field(gt=0, description="Intero positivo")
    my_float: float = Field(gt=0, lt=1, description="Float tra 0 e 1")
    my_str: str = Field(min_length=1, description="Stringa alfanumerica")
    my_list: List[int] = Field(min_length=17, max_length=17, description="Lista di 17 interi")
    my_tuple: Tuple[str, ...] = Field(description="Tupla di stringhe")
    my_series: Series[PositiveFloatSeries] = Field(description="Serie con valori positivi")
    my_dataframe: DataFrame[DatedDataFrame] = Field(description="DataFrame con indice temporale")
    extra_data: List[ExtraDataModel] = Field(min_length=1, description="Lista di metadati")

    class Config:
        arbitrary_types_allowed = True
        validate_assignment = True
        str_strip_whitespace = True

    # Validatore per la stringa
    @field_validator('my_str')
    @classmethod
    def validate_my_str(cls, v):
        if not v.strip():
            raise ValueError('La stringa non può essere vuota')
        if not re.match(r'^[a-zA-Z0-9.]+$', v):
            raise ValueError('La stringa può contenere solo lettere, numeri e il punto')
        return v.strip()
    
    # Validatore per la lista
    @field_validator('my_list')
    @classmethod
    def validate_my_list(cls, v):
        for i, item in enumerate(v):
            if not (5 <= item <= 27):
                raise ValueError(f'Il valore alla posizione {i} ({item}) deve essere compreso tra 5 e 27')
        return v

def process_data(inputs: InputModel) -> None:
    """
    Processa i dati validati e stampa un riepilogo.
    
    Args:
        inputs: Modello InputModel validato
    """
    logger.info("Dati ricevuti e validati con successo")
    print("=" * 50)
    print("✅ DATI VALIDATI CON SUCCESSO")
    print("=" * 50)
    
    print(f"🔢 Intero: {inputs.my_int}")
    print(f"📊 Float: {inputs.my_float}")
    print(f"📝 Stringa: '{inputs.my_str}'")
    print(f"📋 Lista: {len(inputs.my_list)} elementi (range: {min(inputs.my_list)}-{max(inputs.my_list)})")
    print(f"📦 Tupla: {len(inputs.my_tuple)} elementi")
    print(f"📈 Serie: {len(inputs.my_series)} valori (media: {inputs.my_series.mean():.2f})")
    print(f"🗓️ DataFrame: {len(inputs.my_dataframe)} righe, periodo: {inputs.my_dataframe.index.min().date()} - {inputs.my_dataframe.index.max().date()}")
    
    print(f"\n📊 Informazioni aggiuntive ({len(inputs.extra_data)} elementi):")
    for i, data in enumerate(inputs.extra_data):
        duration_days = (data.date_1 - data.date_0).days
        time_span = data.t_1 - data.t_0
        print(f"  📌 Elemento {i+1}: Index={data.index}, Durata={duration_days} giorni, Span={time_span}")

def run_examples():
    """Esegue esempi di validazione con dati validi e invalidi"""
    
    # Esempio 1: Dati validi con Series che rispettano lo schema
    print("\n" + "="*60)
    print("🧪 TEST 1: DATI VALIDI")
    print("="*60)
    
    # Serie principale con indice (opzionale per PositiveFloatSeries)
    valid_series = pd.Series([1.5, 2.0, 5.8], name='serie_valori', dtype=float)
    
    # Series per eq_line - DEVONO rispettare PositiveFloatSeries
    valid_series_list = [
        pd.Series([10.1, 10.2, 15.5], dtype=float, name='sub_series_1'),
        pd.Series([20.1, 20.2, 25.8], dtype=float, name='sub_series_2')
    ]
    
    valid_df = pd.DataFrame(
        {
            'valore_di_interesse': [10.5, 20.8],
            'eq_line': valid_series_list
        },
        index=pd.to_datetime(['2025-09-01', '2025-09-02'])
    )
    
    valid_extra_data_list = [
        {"index": 1, "t_0": 5, "t_1": 10, "date_0": datetime(2025, 1, 1), "date_1": datetime(2025, 1, 10)},
        {"index": 2, "t_0": 10, "t_1": 20, "date_0": datetime(2025, 2, 1), "date_1": datetime(2025, 2, 20)}
    ]
    
    valid_data = {
        "my_int": 1,
        "my_float": 0.5,
        "my_str": 'testo.prova.123',
        "my_list": [10, 15, 20, 25, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18],
        "my_tuple": ('a', 'b', 'c'),
        "my_series": valid_series,
        "my_dataframe": valid_df,
        "extra_data": valid_extra_data_list
    }
    
    try:
        valid_inputs = InputModel(**valid_data)
        process_data(valid_inputs)
    except ValidationError as e:
        print(f"❌ Errore di validazione Pydantic: {e}")
    except pa.errors.SchemaError as e:
        print(f"❌ Errore di schema Pandera: {e}")
    except Exception as e:
        print(f"❌ Errore inatteso: {e}")
    
    # Esempio 2: Dati non validi (valore negativo nella Serie nested)
    print("\n" + "="*60)
    print("🧪 TEST 2: DATI INVALIDI - Serie nested con valore negativo")
    print("="*60)
    
    invalid_series_list = [
        pd.Series([10.1, -2.0, 15.5], dtype=float, name='invalid_series'),  # Valore negativo!
        pd.Series([20.1, 20.2, 25.8], dtype=float, name='valid_series')
    ]
    
    invalid_df = pd.DataFrame(
        {
            'valore_di_interesse': [10.5, 20.8],
            'eq_line': invalid_series_list
        },
        index=pd.to_datetime(['2025-09-01', '2025-09-02'])
    )
    
    invalid_data_1 = {**valid_data, 'my_dataframe': invalid_df}
    
    try:
        InputModel(**invalid_data_1)
        print("❌ ERRORE: I dati invalidi sono stati accettati!")
    except ValidationError as e:
        print("✅ Errore Pydantic correttamente catturato")
        print(f"📝 Dettagli: {str(e).split('Input should be')[0].strip()}")
    except pa.errors.SchemaError as e:
        print("✅ Errore Pandera correttamente catturato")
        print(f"📝 Dettagli: {str(e)[:200]}...")
    except Exception as e:
        print(f"✅ Errore generico catturato: {type(e).__name__}")
    
    # Esempio 3: Dati non validi (t_1 <= t_0)
    print("\n" + "="*60)
    print("🧪 TEST 3: DATI INVALIDI - t_1 <= t_0")
    print("="*60)
    
    invalid_extra_data_list = [
        {"index": 1, "t_0": 5, "t_1": 10, "date_0": datetime(2025, 1, 1), "date_1": datetime(2025, 1, 10)},
        {"index": 2, "t_0": 20, "t_1": 10, "date_0": datetime(2025, 2, 1), "date_1": datetime(2025, 2, 20)}  # t_1 < t_0
    ]
    invalid_data_2 = {**valid_data, 'extra_data': invalid_extra_data_list}
    
    try:
        InputModel(**invalid_data_2)
        print("❌ ERRORE: I dati invalidi sono stati accettati!")
    except ValidationError as e:
        print("✅ Errore di validazione correttamente catturato")
        print(f"📝 Dettagli: {str(e).split('Input should be')[0].strip()}")
    except Exception as e:
        print(f"✅ Errore generico catturato: {type(e).__name__}")

if __name__ == "__main__":
    run_examples()