تنظيم خطوات التحويل في صورة دوال

# Notebook-style (fragile)
df = pd.read_csv('orders.csv')
df = df.dropna(subset=['revenue'])
df = df[df['quantity'] > 0]
df['revenue_per_unit'] = df['revenue'] / df['quantity']

# Function-style (robust)
def extract(path):
    return pd.read_csv(path)

def transform(df):
    return (
        df.dropna(subset=['revenue'])
        .query('quantity > 0')
        .assign(revenue_per_unit=lambda d: d['revenue'] / d['quantity'])
    )

df = transform(extract('orders.csv'))

def extract(config):
    return pd.read_csv(config['input_path'], parse_dates=['order_date'])

def transform(df, config):
    return (
        df
        .dropna(subset=config['required_cols'])
        .query('quantity > 0')
        .assign(revenue=lambda d: d['quantity'] * d['unit_price'])
    )

def load(df, config):
    df.to_parquet(config['output_path'], index=False)
    print(f'Saved {len(df)} rows.')

def drop_null_rows(df, required_cols):
    return df.dropna(subset=required_cols)

def remove_returns(df):
    return df[df['quantity'] > 0]

def compute_revenue(df):
    return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])

def add_date_features(df):
    return df.assign(
        year=lambda d: d['order_date'].dt.year,
        month=lambda d: d['order_date'].dt.month
    )

import pandas as pd

REQUIRED = ['order_id', 'revenue', 'order_date']

def transform(raw_df):
    return (
        raw_df
        .pipe(drop_null_rows, required_cols=REQUIRED)
        .pipe(remove_returns)
        .pipe(compute_revenue)
        .pipe(add_date_features)
    )

df_clean = transform(pd.read_csv('orders.csv', parse_dates=['order_date']))
print(df_clean.shape)

audit_log = []

def audited(func):
    def wrapper(df, *args, **kwargs):
        before = len(df)
        result = func(df, *args, **kwargs)
        after = len(result)
        audit_log.append({'step': func.__name__, 'in': before, 'out': after, 'dropped': before - after})
        return result
    return wrapper

@audited
def drop_null_rows(df, required_cols):
    return df.dropna(subset=required_cols)

import pandas as pd

def test_drop_null_rows():
    test_df = pd.DataFrame({
        'order_id': [1, 2, 3],
        'revenue': [100.0, None, 200.0]
    })
    result = drop_null_rows(test_df, required_cols=['revenue'])
    assert len(result) == 2, 'Should have 2 non-null rows'
    assert result['revenue'].isna().sum() == 0, 'No nulls in revenue'
    print('test_drop_null_rows PASSED')

test_drop_null_rows()

# pipeline.py
# from extract import extract_orders
# from transform import transform
# from load import load_to_parquet
# from validate import validate_sales_df

# def run_pipeline(config):
#     raw = extract_orders(config)
#     clean = transform(raw, config)
#     validate_sales_df(clean)
#     load_to_parquet(clean, config)

print('Module-based pipeline structure shown above (imports commented for demo)')

def add_revenue_flag(df, threshold=500):
    # Use copy to avoid mutating the input
    df = df.copy()
    df['is_large_order'] = df['revenue'] >= threshold
    return df

# Running twice gives the same result
df1 = add_revenue_flag(df_clean)
df2 = add_revenue_flag(df_clean)
assert df1.equals(df2), 'Function is not idempotent!'
print('Idempotency check passed.')

from typing import List

def drop_null_rows(df: pd.DataFrame, required_cols: List[str]) -> pd.DataFrame:
    return df.dropna(subset=required_cols)

def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
    return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])

print('Type-annotated functions ready for production use.')

def compute_revenue(df: pd.DataFrame) -> pd.DataFrame:
    """Add 'revenue' column as quantity * unit_price.

    Requires: 'quantity' (numeric), 'unit_price' (numeric) columns.
    Returns: df with new 'revenue' float column appended.
    """
    return df.assign(revenue=lambda d: d['quantity'] * d['unit_price'])

help(compute_revenue)

import time

CONFIG = {
    'input_path': 'orders.csv',
    'output_path': 'orders_clean.parquet',
    'required_cols': ['order_id', 'revenue']
}

t0 = time.time()
raw = extract(CONFIG)
clean = transform(raw, CONFIG)
load(clean, CONFIG)

print(f'Pipeline completed in {time.time()-t0:.1f}s')
print(f'Audit log: {audit_log}')