Saját adatvalidációs DSL Pythonban

Az adatvalidációs kódok karbantartása gyakran rémálom. Az üzleti szabályok elvesznek a beágyazott feltételek dzsungelében, a hibaellenőrzés pedig összekeveredik a logikával. Új szabály hozzáadása sokszor azt jelenti, hogy procedurális függvényekben kell kutatni a megfelelő hely után. Bár léteznek keretrendszerek, most egy saját, pehelykönnyű megoldást építünk Pythonban.

Mi is az a DSL?

Írjunk egy egyszerű tartományspecifikus nyelvet (DSL – Domain-Specific Language). Ez egy olyan szókészletet ad a kezünkbe, amely kifejezetten az adatellenőrzésre született. Ahelyett, hogy általános Python kódot írnánk, olyan függvényeket és osztályokat hozunk létre, amelyek tükrözik a gondolkodásmódunkat.

Az adatvalidációnál ez azt jelenti, hogy a szabályok úgy olvashatók, mint az üzleti követelmények:

• „az életkor 18 és 120 között legyen” vagy
• „az e-mail tartalmazzon @ jelet”.

A DSL elvégzi a technikai munkát, mi pedig a lényegre koncentrálhatunk. Az eredmény egy olvasható, tesztelhető és bővíthető kód. Vágjunk is bele!

Miért építsünk saját DSL-t?

Nézzük meg, hogyan néz ki egy tipikus adatellenőrzés Pythonban:

def validate_customers(df):
    errors = []
    if df['customer_id'].duplicated().any():
        errors.append("Duplicate IDs")
    if (df['age'] < 0).any():
        errors.append("Negative ages")
    if not df['email'].str.contains('@').all():
        errors.append("Invalid emails")
    return errors

Ez a megközelítés beégeti a logikát, keveri az üzleti szabályokat a hibakezeléssel, és karbantarthatatlanná válik. Ezzel szemben a DSL szétválasztja ezeket a feladatokat. A procedurális függvények helyett így fog kinézni a kódunk:

# Hagyományos megközelítés
if df['age'].min() < 0 or df['age'].max() > 120:
    raise ValueError("Invalid ages found")

# DSL megközelítés
validator.add_rule(Rule("Valid ages", between('age', 0, 120), "Ages must be 0-120"))

Ez a módszer elválasztja azt, amit ellenőrzünk (üzleti szabály), attól, ahogyan a hibákat kezeljük (jelentés).

Mintaadatkészlet létrehozása

Kezdésként hozzunk létre egy valósághű e-kereskedelmi adatkészletet, tele gyakori hibákkal:

import pandas as pd

customers = pd.DataFrame({
    'customer_id': [101, 102, 103, 103, 105],
    'email': ['john@gmail.com', 'invalid-email', '', 'sarah@yahoo.com', 'mike@domain.co'],
    'age': [25, -5, 35, 200, 28],
    'total_spent': [250.50, 1200.00, 0.00, -50.00, 899.99],
    'join_date': ['2023-01-15', '2023-13-45', '2023-02-20', '2023-02-20', '']
}) # Megjegyzés: a 2023-13-45 szándékosan hibás dátum.

Ez az adathalmaz tartalmaz duplikált azonosítókat, rossz e-mail formátumokat, lehetetlen életkorokat és negatív költéseket. Tökéletes tesztpálya.

A validációs logika megírása

A Rule osztály létrehozása

Készítsünk egy Rule osztályt, amely becsomagolja a validációs logikát:

class Rule:
    def __init__(self, name, condition, error_msg):
        self.name = name
        self.condition = condition
        self.error_msg = error_msg
    
    def check(self, df):
        # A feltétel függvény True értéket ad a HELYES sorokra.
        # A ~ (bitenkénti NOT) segítségével kiválasztjuk a HIBÁS sorokat.
        violations = df[~self.condition(df)]
        if not violations.empty:
            return {
                'rule': self.name,
                'message': self.error_msg,
                'violations': len(violations),
                'sample_rows': violations.head(3).index.tolist()
            }
        return None

A condition paraméter bármilyen olyan függvény lehet, amely egy DataFrame-et vár, és egy logikai sorozatot (Boolean Series) ad vissza. A check metódus részletes jelentést ad a hibákról, beleértve a szabály nevét és a mintasorokat a hibakereséshez.

Több szabály hozzáadása

Most írjuk meg a DataValidator osztályt, amely kezeli a szabályok gyűjteményét:

class DataValidator:
    def __init__(self):
        self.rules = []
    
    def add_rule(self, rule):
        self.rules.append(rule)
        return self # Lehetővé teszi a metódusláncolást
    
    def validate(self, df):
        results = []
        for rule in self.rules:
            violation = rule.check(df)
            if violation:
                results.append(violation)
        return results

Az add_rule metódus self-et ad vissza, így láncolhatjuk a hívásokat. A validate metódus minden szabályt függetlenül futtat le, így egy hiba nem állítja meg a folyamatot.

Olvasható feltételek építése

A szabályokhoz szükségünk van olvasható segédfüggvényekre. A lambda függvények működnek, de nehezen olvashatók. Írjunk inkább beszédesebb függvényeket:

def not_null(column):
    return lambda df: df[column].notna()

def unique_values(column):
    return lambda df: ~df.duplicated(subset=[column], keep=False)

def between(column, min_val, max_val):
    return lambda df: df[column].between(min_val, max_val)

Ezek a segédfüggvények lambda kifejezéseket adnak vissza, amelyek a pandas logikai műveleteivel dolgoznak. A minták illesztéséhez a reguláris kifejezések is egyszerűvé válnak:

import re

def matches_pattern(column, pattern):
    return lambda df: df[column].str.match(pattern, na=False)

A na=False paraméter biztosítja, hogy a hiányzó értékek hibának minősüljenek.

Validátor építése a mintaadatokhoz

Nézzük meg, hogyan működik a DSL a gyakorlatban:

validator = DataValidator()

validator.add_rule(Rule(
   "Unique customer IDs", 
   unique_values('customer_id'),
   "Customer IDs must be unique across all records"
))

validator.add_rule(Rule(
   "Valid email format",
   matches_pattern('email', r'^[^@\s]+@[^@\s]+\.[^@\s]+$'),
   "Email addresses must contain @ symbol and domain"
))

validator.add_rule(Rule(
   "Reasonable customer age",
   between('age', 13, 120),
   "Customer age must be between 13 and 120 years"
))

validator.add_rule(Rule(
   "Non-negative spending",
   lambda df: df['total_spent'] >= 0,
   "Total spending amount cannot be negative"
))

Minden szabály ugyanazt a mintát követi: név, feltétel, hibaüzenet. A kód szinte üzleti követelményként olvasható. Most futtassuk le az ellenőrzést:

issues = validator.validate(customers)

for issue in issues:
    print(f"❌ Rule: {issue['rule']}")
    print(f"Problem: {issue['message']}")
    print(f"Affected rows: {issue['sample_rows']}")
    print()

A kimenet egyértelműen azonosítja a problémákat és azok helyét.

Oszlopokon átívelő validációk

A valós szabályok gyakran több oszlopot érintenek. Erre is van megoldás:

def high_spender_email_required(df):
    high_spenders = df['total_spent'] > 500
    has_valid_email = df['email'].str.contains('@', na=False)
    # Érvényes, ha: (NEM nagy költő) VAGY (Van érvényes e-mailje)
    return ~high_spenders | has_valid_email

validator.add_rule(Rule(
    "High Spenders Need Valid Email",
    high_spender_email_required,
    "Customers spending over $500 must have valid email addresses"
))

Ez a szabály Boole-logikát használ: a sokat költő vevőknek kötelező az e-mail, a többieknek nem.

Dátumok kezelése

A dátumok validálása körültekintést igényel a formátumhibák miatt:

def valid_date_format(column, date_format='%Y-%m-%d'):
    def check_dates(df):
        # Az errors='coerce' a hibás dátumokat NaT-ra alakítja
        parsed_dates = pd.to_datetime(df[column], format=date_format, errors='coerce')
        # Érvényes, ha az eredeti nem null ÉS a parse-olt sem NaT
        return df[column].notna() & parsed_dates.notna()
    return check_dates

validator.add_rule(Rule(
    "Valid Join Dates",
    valid_date_format('join_date'),
    "Join dates must follow YYYY-MM-DD format"
))

Itt az errors=’coerce’ paraméter segít elegánsan kezelni a rossz formátumokat.

Dekorátor minta a folyamatokhoz

Éles környezetben dekorátorokkal integrálhatjuk az ellenőrzést:

def validate_dataframe(validator):
    def decorator(func):
        def wrapper(df, *args, **kwargs):
            issues = validator.validate(df)
            if issues:
                error_details = [f"{issue['rule']}: {issue['violations']} violations" for issue in issues]
                raise ValueError(f"Data validation failed: {'; '.join(error_details)}")
            return func(df, *args, **kwargs)
        return wrapper
    return decorator

# Feltételezve, hogy a 'customer_validator' már létezik
# @validate_dataframe(customer_validator)
def process_customer_data(df):
    return df.groupby('age').agg({'total_spent': 'sum'})

Ez biztosítja, hogy a hibás adatok ne kerüljenek be a feldolgozási folyamatba.

A rendszer bővítése

A DSL könnyen kiegészíthető további logikával:

# Statisztikai kiugró értékek keresése
def within_standard_deviations(column, std_devs=3):
    return lambda df: abs(df[column] - df[column].mean()) <= std_devs * df[column].std()

# Referencia integritás ellenőrzése
def foreign_key_exists(column, reference_df, reference_column):
    return lambda df: df[column].isin(reference_df[reference_column])

# Egyedi üzleti logika
def profit_margin_reasonable(df):
    margin = (df['revenue'] - df['cost']) / df['revenue']
    return (margin >= 0) & (margin <= 1)

Ezek a függvények mind kompatibilisek a rendszerünkkel és egyszerűen beilleszthetők a validátorba.