refactor: remove code overhead and redundant logging

- Remove verbose 'Received /X request' logs from all 11 route handlers
- Remove redundant Arrow conversion logs from logic modules
- Remove all logging from arrow_utils.py (low-level serialization noise)
- Remove unused logger/logging imports from 3 extract logic modules
- Remove unnecessary comments from join.py, extract modules
- Remove dead _has_request_context() wrapper in service_utils.py
- Remove dead metadata field from StepResult dataclass
- Remove unreachable else branch and per-format logs in load.py
- Remove decorative banner comments from service_utils.py
- Fix deprecated datetime.utcnow() in xcom_file_utils.py
- Remove unused pyarrow import from columns.py
- Fix variable shadowing (l -> layer) in test_pipeline_compiler.py
- Add pyyaml to CI dependencies
- All 208 tests passing

Author: VTvito

.github/workflows/ci.yml

@@ -34,7 +34,7 @@ jobs:
       - name: Install test dependencies
         run: |
           pip install pytest pytest-cov
-          pip install flask pyarrow pandas numpy prometheus_client xlsxwriter openpyxl requests
+          pip install flask pyarrow pandas numpy prometheus_client xlsxwriter openpyxl requests pyyaml
       - name: Run unit tests
         run: pytest tests/unit/ -v --tb=short --cov=services --cov-report=xml
       - name: Upload coverage

ai_agent/pipeline_compiler.py

@@ -34,7 +34,6 @@ class StepResult:
     duration_sec: float = 0.0
     data_size_bytes: int = 0
     error_message: str = ""
-    metadata: dict = field(default_factory=dict)
 
 
 @dataclass

airflow/dags/xcom_file_utils.py

@@ -11,7 +11,7 @@
 import logging
 import os
 import uuid
-from datetime import datetime
+from datetime import datetime, timezone
 
 logger = logging.getLogger("xcom_file_utils")
 
@@ -33,7 +33,7 @@ def save_ipc_to_shared(ipc_data: bytes, dataset_name: str, step_name: str) -> st
     xcom_dir = os.path.join(SHARED_DATA_ROOT, dataset_name, "xcom")
     os.makedirs(xcom_dir, exist_ok=True)
 
-    timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
+    timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ")
     unique_id = uuid.uuid4().hex[:8]
     filename = f"{step_name}_{timestamp}_{unique_id}.arrow"
     file_path = os.path.join(xcom_dir, filename)

services/clean-nan-service/app/clean.py

@@ -33,7 +33,6 @@ def apply_transformations(arrow_table, strategy="drop", fill_value=None, columns
 
     try:
         df = arrow_table.to_pandas()
-        logger.info(f"Converted to Pandas with shape {df.shape}, strategy='{strategy}'")
 
         total_cells = df.size
         total_null_before = int(df.isna().sum().sum())
@@ -74,11 +73,6 @@ def apply_transformations(arrow_table, strategy="drop", fill_value=None, columns
         total_null_after = int(df_cleaned.isna().sum().sum())
         nulls_handled = total_null_before - total_null_after
 
-        logger.info(
-            f"Strategy '{strategy}': handled {nulls_handled} nulls. "
-            f"Rows {arrow_table.num_rows} -> {df_cleaned.shape[0]}"
-        )
-
         cleaned_arrow_table = pa.Table.from_pandas(df_cleaned)
         return (cleaned_arrow_table, nulls_handled, total_null_before, total_cells)
     except Exception as e:

services/clean-nan-service/app/routes.py

@@ -28,7 +28,6 @@ def clean_nan():
     correlation_id = get_correlation_id()
     try:
         REQUEST_COUNTER.inc()
-        logger.info("Received /clean-nan request.", extra={"correlation_id": correlation_id})
 
         params = parse_x_params()
         dataset_name = params.get('dataset_name')

services/common/arrow_utils.py

@@ -1,22 +1,16 @@
-import logging
-
 import pyarrow as pa
 import pyarrow.ipc as pa_ipc
 
-logger = logging.getLogger(__name__)
 
 def ipc_to_table(ipc_data: bytes) -> pa.Table:
     """
     Deserialize bytes of Arrow IPC in Arrow Table.
     """
     try:
         reader = pa_ipc.open_stream(pa.BufferReader(ipc_data))
-        table = reader.read_all()
-        logger.info(f"Deserialized Arrow Table with {table.num_rows} rows and {table.num_columns} columns.")
-        return table
+        return reader.read_all()
     except Exception as e:
-        logger.error(f"Failed to parse Arrow IPC data: {e}")
-        raise
+        raise ValueError(f"Failed to parse Arrow IPC data: {e}") from e
 
 def table_to_ipc(table: pa.Table) -> bytes:
     """
@@ -26,9 +20,6 @@ def table_to_ipc(table: pa.Table) -> bytes:
         sink = pa.BufferOutputStream()
         with pa_ipc.new_stream(sink, table.schema) as writer:
             writer.write_table(table)
-        ipc_bytes = sink.getvalue().to_pybytes()
-        logger.info(f"Serialized Arrow Table to IPC format, {len(ipc_bytes)} bytes.")
-        return ipc_bytes
+        return sink.getvalue().to_pybytes()
     except Exception as e:
-        logger.error(f"Failed to serialize Arrow Table to IPC format: {e}")
-        raise
+        raise ValueError(f"Failed to serialize Arrow Table to IPC: {e}") from e

services/common/service_utils.py

@@ -23,7 +23,6 @@
 from flask import Response, g, has_request_context, jsonify, request
 from prometheus_client import Counter, generate_latest
 
-# ── Prometheus helpers ──────────────────────────────────────────────
 
 def create_service_counters(service_slug):
     """
@@ -42,7 +41,6 @@ def create_service_counters(service_slug):
     )
 
 
-# ── Standard endpoints ─────────────────────────────────────────────
 
 def register_standard_endpoints(bp, service_name):
     """
@@ -62,7 +60,6 @@ def metrics():
         return Response(generate_latest(), mimetype="text/plain")
 
 
-# ── Request helpers ─────────────────────────────────────────────────
 
 def get_correlation_id():
     """
@@ -92,7 +89,6 @@ def parse_x_params():
         raise ValueError(f"Malformed JSON in X-Params header: {exc}") from exc
 
 
-# ── Metadata helper ────────────────────────────────────────────────
 
 def save_metadata(service_name, dataset_name, extra_fields=None, start_time=None):
     """
@@ -118,7 +114,7 @@ def save_metadata(service_name, dataset_name, extra_fields=None, start_time=None
         "service_name": service_name,
         "dataset_name": dataset_name,
         "timestamp": timestamp,
-        "correlation_id": get_correlation_id() if _has_request_context() else None,
+        "correlation_id": get_correlation_id() if has_request_context() else None,
     }
     if start_time is not None:
         metadata["duration_sec"] = round(time.time() - start_time, 3)
@@ -130,7 +126,3 @@ def save_metadata(service_name, dataset_name, extra_fields=None, start_time=None
 
     return metadata_path
 
-
-def _has_request_context():
-    """Check whether we're inside a Flask request context."""
-    return has_request_context()

services/data-quality-service/app/routes.py

@@ -52,7 +52,6 @@ def data_quality():
     correlation_id = get_correlation_id()
     try:
         REQUEST_COUNTER.inc()
-        logger.info("Received /data-quality request.", extra={"correlation_id": correlation_id})
 
         header_data = parse_x_params()
         dataset_name = header_data.get('dataset_name')

services/delete-columns-service/app/columns.py

@@ -1,7 +1,5 @@
 import logging
 
-import pyarrow as pa
-
 logger = logging.getLogger('delete-columns-service')
 
 def drop_columns_arrow(arrow_table, columns_to_delete):

services/delete-columns-service/app/routes.py

@@ -28,7 +28,6 @@ def delete_columns():
     correlation_id = get_correlation_id()
     try:
         REQUEST_COUNTER.inc()
-        logger.info("Received /delete-columns request.", extra={"correlation_id": correlation_id})
 
         header_data = parse_x_params()
         columns_raw = header_data.get('columns', '')