Skip to content

Commit 4476d79

Browse files
committed
feat(pipeline): add in-memory DataFrame support and refactor app.py SSOT
1 parent 10fd6e0 commit 4476d79

2 files changed

Lines changed: 35 additions & 29 deletions

File tree

‎app.py‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,10 +19,8 @@
1919
sys.path.insert(0, str(Path(__file__).parent))
2020

2121
from config.settings import CLEANING_CONFIG, PIPELINE_VERSION, VALIDATION_CONFIG
22-
from src.cleaning import clean_data
22+
from src.pipeline import execute_pipeline
2323
from src.report import generate_quality_report
24-
from src.transformation import transform_data
25-
from src.validation import validate_data
2624

2725
# ── Page config ───────────────────────────────────────────────────────────────
2826
st.set_page_config(
@@ -143,11 +141,14 @@ def run_pipeline_cached(df_raw: pd.DataFrame, normalize: bool, null_thresh: floa
143141
"duplicate_threshold_pct": dup_thresh,
144142
"iqr_factor": 1.5,
145143
}
146-
before_report = validate_data(df_raw, **v_config)
147-
df_clean = clean_data(df_raw, config=CLEANING_CONFIG)
148-
df_final = transform_data(df_clean, normalize=normalize)
149-
after_report = validate_data(df_final)
150-
return before_report, df_clean, df_final, after_report
144+
result = execute_pipeline(
145+
df_raw=df_raw,
146+
validation_config=v_config,
147+
normalize=normalize,
148+
)
149+
if not result.success:
150+
raise RuntimeError(result.error_message or "Error executing pipeline")
151+
return result.before_report, result.df_clean, result.df_final, result.after_report
151152

152153

153154
with st.spinner("Ejecutando pipeline..."):

‎src/pipeline.py‎

Lines changed: 26 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -59,10 +59,11 @@ def execute_pipeline(
5959
cleaning_config: dict[str, Any] | None = None,
6060
validation_config: dict[str, Any] | None = None,
6161
normalize: bool = False,
62+
df_raw: pd.DataFrame | None = None,
6263
) -> PipelineResult:
6364
"""
6465
Execute the unified 5-step DataPrep Pipeline:
65-
1. Ingestion
66+
1. Ingestion (skipped if df_raw is provided)
6667
2. Validation (Before)
6768
3. Cleaning
6869
4. Transformation
@@ -77,6 +78,7 @@ def execute_pipeline(
7778
cleaning_config: Optional cleaning settings dict override.
7879
validation_config: Optional validation threshold dict override.
7980
normalize: Whether to apply min-max normalization to numeric columns.
81+
df_raw: Optional in-memory raw DataFrame to bypass file ingestion.
8082
8183
Returns:
8284
PipelineResult with execution status, DataFrames, reports, and execution time.
@@ -95,26 +97,29 @@ def execute_pipeline(
9597

9698
# ── STEP 1: INGEST ──────────────────────────────────────────────────────
9799
logger.info("[1/5] INGESTION")
98-
try:
99-
if file_format == "excel":
100-
df_raw = load_excel(input_path)
101-
elif file_format == "api":
102-
if not api_url:
103-
err_msg = "API URL required when format is 'api'"
104-
logger.error(err_msg)
105-
return PipelineResult(success=False, error_message=err_msg)
106-
df_raw = load_json_api(api_url)
107-
else:
108-
df_raw = load_csv(input_path)
109-
except FileNotFoundError as e:
110-
err_msg = f"Input file not found: {e}"
111-
logger.error(err_msg)
112-
logger.error("Run 'python generate_dataset.py' first to create test data.")
113-
return PipelineResult(success=False, error_message=err_msg)
114-
except Exception as e:
115-
err_msg = f"Ingestion error: {e}"
116-
logger.error(err_msg, exc_info=True)
117-
return PipelineResult(success=False, error_message=err_msg)
100+
if df_raw is None:
101+
try:
102+
if file_format == "excel":
103+
df_raw = load_excel(input_path)
104+
elif file_format == "api":
105+
if not api_url:
106+
err_msg = "API URL required when format is 'api'"
107+
logger.error(err_msg)
108+
return PipelineResult(success=False, error_message=err_msg)
109+
df_raw = load_json_api(api_url)
110+
else:
111+
df_raw = load_csv(input_path)
112+
except FileNotFoundError as e:
113+
err_msg = f"Input file not found: {e}"
114+
logger.error(err_msg)
115+
logger.error("Run 'python generate_dataset.py' first to create test data.")
116+
return PipelineResult(success=False, error_message=err_msg)
117+
except Exception as e:
118+
err_msg = f"Ingestion error: {e}"
119+
logger.error(err_msg, exc_info=True)
120+
return PipelineResult(success=False, error_message=err_msg)
121+
else:
122+
logger.info(f"Using in-memory DataFrame ({len(df_raw):,} rows)")
118123

119124
try:
120125
# ── STEP 2: VALIDATE (BEFORE) ─────────────────────────────────────────

0 commit comments

Comments
 (0)