docs: refactor pipeline documentation to design-contract architecture

- Decoupled stage Executor and Logic into distinct contracts
- Map component-level failure surfaces and severity models
- Align Global Orchestrator docstrings with the Promote-Purge-Restore Cloud-Sync pattern

Author: BLMgithub

data_extract/run_extract.py

@@ -20,6 +20,7 @@
 MIME_TYPE = "application/vnd.google-apps.folder"
 
 
+# TODO: wrap steps into functions
 def run_extraction(target_child_folder):
 
     service = get_drive_service()

data_pipeline/assembly/assembly_executor.py

@@ -43,11 +43,11 @@ def orchestrate_event_assembly(run_context: RunContext, report: Dict) -> bool:
     Coordinates the linear transformation pipeline for order-grain events.
 
     Execution Flow:
-    1. Load: Fetch 'orders', 'items', and 'payments'.
-    2. Merge: Join into a single row-per-order grain.
-    3. Derive: Calculate analytical time-deltas and lineage.
-    4. Freeze: Enforce semantic schema and dtypes.
-    5. Export: Persist to the assembly zone.
+        1. Load: Fetch 'orders', 'items', and 'payments'.
+        2. Merge: Join into a single row-per-order grain.
+        3. Derive: Calculate analytical time-deltas and lineage.
+        4. Freeze: Enforce semantic schema and dtypes.
+        5. Export: Persist to the assembly zone.
 
     Memory Management:
     - Explicitly deletes intermediate DataFrames and triggers gc.collect()
@@ -152,16 +152,16 @@ def assemble_events(run_context: RunContext) -> dict:
     tables into contract-compliant analytical datasets.
 
     Workflow I: Event Assembly (Order Grain)
-    1. Load: Fetches core event tables (Orders, Items, Payments).
-    2. Merge: Join datasets with strict 1:1 order_id cardinality enforcement.
-    3. Derive: Calculate temporal metrics (lead times) and lineage attributes.
-    4. Freeze: Project final schema and enforce strictly defined dtypes.
-    5. Export: Persist the unified event table to the Gold zone.
+        1. Load: Fetches core event tables (Orders, Items, Payments).
+        2. Merge: Join datasets with strict 1:1 order_id cardinality enforcement.
+        3. Derive: Calculate temporal metrics (lead times) and lineage attributes.
+        4. Freeze: Project final schema and enforce strictly defined dtypes.
+        5. Export: Persist the unified event table to the Gold zone.
 
     Workflow II: Dimension Reference Extraction
-    1. Iterate: Process Customer and Product registries.
-    2. Extract: Select required columns and deduplicate by primary key.
-    3. Export: Persist independent reference tables to the Gold zone.
+        1. Iterate: Process Customer and Product registries.
+        2. Extract: Select required columns and deduplicate by primary key.
+        3. Export: Persist independent reference tables to the Gold zone.
 
     Operational Guarantees:
     - Grain: Strictly one row per 'order_id' for the event dataset.

data_pipeline/contract/contract_executor.py

@@ -22,13 +22,13 @@ def apply_contract(
     ensuring only compliant rows and columns reach the Silver (contracted) layer.
 
     Workflow:
-    1. Resolve: Determines table configuration and role (event_fact, entity_reference, etc.).
-    2. Load: Fetches the raw snapshot from the lake's snapshot zone.
-    3. Sequence: Iteratively applies atomic filtering rules (Deduplication, Null-checks, etc.).
-    4. Track: Captures row-level telemetry and identifies compromised 'order_id's.
-    5. Propagate: Returns validated/invalidated IDs to maintain referential integrity.
-    6. Freeze: Executes 'enforce_schema' as the terminal step to project approved columns.
-    7. Export: Persists the contract-compliant dataset to the Silver zone.
+        1. Resolve: Determines table configuration and role (event_fact, entity_reference, etc.).
+        2. Load: Fetches the raw snapshot from the lake's snapshot zone.
+        3. Sequence: Iteratively applies atomic filtering rules (Deduplication, Null-checks, etc.).
+        4. Track: Captures row-level telemetry and identifies compromised 'order_id's.
+        5. Propagate: Returns validated/invalidated IDs to maintain referential integrity.
+        6. Freeze: Executes 'enforce_schema' as the terminal step to project approved columns.
+        7. Export: Persists the contract-compliant dataset to the Silver zone.
 
     Operational Guarantees:
     - Subtractive Only: Filters rows first; never mutates row values (only column types).

data_pipeline/publish/publish_executor.py

@@ -20,12 +20,12 @@ def execute_publish_lifecycle(run_context: RunContext) -> Dict:
     the internal assembly zones to the production-facing BI environment.
 
     Workflow:
-    1. Integrity Gate: Verifies that the current run has produced all
-       required semantic modules and tables defined in the registry.
-    2. Promotion: Moves/copies artifacts into a permanent, read-only
-       versioned directory (v{run_id}).
-    3. Activation: Performs an atomic update of the 'latest' pointer
-       to switch BI/Reporting traffic to the new version.
+        1. Integrity Gate: Verifies that the current run has produced all
+        required semantic modules and tables defined in the registry.
+        2. Promotion: Moves/copies artifacts into a permanent, read-only
+        versioned directory (v{run_id}).
+        3. Activation: Performs an atomic update of the 'latest' pointer
+        to switch BI/Reporting traffic to the new version.
 
     Operational Guarantees:
     - Atomicity: The 'latest' pointer is updated ONLY if all prior

data_pipeline/run_pipeline.py

@@ -33,27 +33,39 @@
 
 def persist_json(path: Path, payload: dict) -> None:
     """
-    Writes the stage report to a JSON file.
+    Serializes state dictionaries to the local filesystem.
+
+    Contract:
+    - Creates parent directories if they do not exist.
+    - Performs standard JSON serialization with 2-space indentation.
     """
     path.parent.mkdir(parents=True, exist_ok=True)
     with open(path, "w") as f:
         json.dump(payload, f, indent=2)
 
 
 def stage_logger(run_context: RunContext, stage: str, report: dict | list):
+    """
+    Standardizes the persistence of stage-level telemetry.
+
+    Contract:
+    - Maps 'stage' identifiers to deterministic filenames (e.g., stage_report.json).
+    - Persists reports to the 'run_context.run_path'.
+    """
 
     persist_json(
         path=run_context.logs_path / f"{stage}.json",
         payload={"run_id": run_context.run_id, "report": report},
     )
 
 
-def initiliaze_metadata(run_context: RunContext) -> None:
+def initialize_metadata(run_context: RunContext) -> None:
     """
-    Run metadata initializer.
+    Registers the commencement of a pipeline run.
 
-    Creates the run-scoped metadata record at pipeline start to
-    establish lifecycle tracking and publish eligibility state.
+    Contract:
+    - Generates a 'run_metadata.json' artifact.
+    - Captures the initial 'run_id', 'start_time', and sets status to 'RUNNING'.
     """
 
     run_dt = dt.strptime(run_context.run_id[:15], "%Y%m%dT%H%M%S")
@@ -65,7 +77,7 @@ def initiliaze_metadata(run_context: RunContext) -> None:
         "started_at": dt.utcnow().isoformat(),
         "run_year": run_dt.year,
         "run_month": run_dt.month,
-        "run_week_of_month": (run_dt.day - 1) // 7 + 1,
+        "run_day": run_dt.day,
         "completed_at": None,
         "run_duration": None,
         "published": False,
@@ -76,10 +88,11 @@ def initiliaze_metadata(run_context: RunContext) -> None:
 
 def finalize_metadata(run_context: RunContext, status: str) -> None:
     """
-    Run metadata finalizer.
+    Updates the run metadata record with terminal status and completion timestamp.
 
-    Updates the run metadata record with terminal status and
-    completion timestamp.
+    Contract:
+    - Updates 'run_metadata.json' with the 'end_time' and final 'status'.
+    - Calculates the total run duration and display in '00m 00s' format.
     """
 
     if not run_context.metadata_path.exists():
@@ -88,18 +101,17 @@ def finalize_metadata(run_context: RunContext, status: str) -> None:
     with open(run_context.metadata_path, "r") as file:
         payload = json.load(file)
 
+    start_time = dt.fromisoformat(payload["started_at"])
     completion_time = dt.utcnow()
 
     payload["status"] = status
     payload["completed_at"] = completion_time.isoformat()
 
-    start_time = dt.fromisoformat(payload["started_at"])
     duration = completion_time - start_time
-
     mm, ss = divmod(int(duration.total_seconds()), 60)
-    payload["run_duration"] = f"{mm:02d}m {ss:02d}s"
 
-    payload["published"] = True if status == "SUCCESS" else False
+    payload["run_duration"] = f"{mm:02d}m {ss:02d}s"
+    payload["published"] = status == "SUCCESS"
 
     persist_json(run_context.metadata_path, payload)
 
@@ -186,32 +198,29 @@ def run_prepublishing_validation_stage(run_context) -> None:
 
 def main() -> None:
     """
-    Pipeline orchestrator.
-
-    Execution order:
-    1. Snapshot raw data
-    2. Initialize metadata (RUNNING)
-    3. Validation
-    4. Contract enforcement
-        - Apply role-driven repair
-        - Propagate invalid order_ids (parent → child)
-        - Propogate valida order_ids (parent → child)
-    5. Re-validation
-    6. Assemble event table
-    7. Build semantic layer
-    8. Pre-publish integrity gate
-    9. Promote version
-    10. Finalize metadata (SUCCESS)
-    11. Atomic activation
-
-    Guarantees:
-    - Deterministic forward-only execution
-    - Single run isolation
-    - Only Contract stage mutates data
-    - Activation occurs only after SUCCESS
-
-    Failure behavior:
-    - Any stage failure → metadata FAILED → process exits
+    Ultimate authority for the end-to-end data pipeline lifecycle.
+
+    Workflow:
+        1.  Initialization: Resolve RunContext and instantiate global run metadata.
+        2.  Ingestion: Synchronize the raw data snapshot from Cloud Storage to local workspace.
+        3.  Gate I (Validation): Assert raw data sanity; fail-fast on fatal structural errors.
+        4.  Processing (Contract): Execute subtractive filtering and Silver-layer schema freezing.
+        5.  Gate II (Revalidation): Defensive check to ensure 'contracted' data is valid.
+        6.  Persistence (Sync Upload): Promote local contracted artifacts to the Cloud Silver Storage.
+        7.  Resource Reclamation: Purge transient directories (raw/contracted) to optimize memory.
+        8.  Hydration (Sync Download): Restore local environment with the full accumulated Silver state.
+        9.  Integration (Assembly): Flatten relational data into a unified Gold event layer.
+        10. Modeling (Semantic): Build entity-centric analytical modules (Fact/Dim).
+        11. Gate III (Pre-Publish): Final verification of semantic artifact completeness.
+        12. Activation (Publish): Atomic swap of the production 'latest' version pointer.
+        13. Finalization: Persist all telemetry/logs to Cloud and purge the local workspace.
+
+    Operational Guarantees:
+    - Defensive Integrity: No data moves to 'Assembly' without passing 'Revalidation'.
+    - Silver Continuity: Uses a Cloud-Sync loop to ensure Assembly operates on the full delta state.
+    - Resource Stewardship: Mandatory local cleanup via global 'finally' block to prevent disk leaks.
+    - Traceability: Enforces atomic 'run_id' consistency across all 13 lifecycle steps.
+    - Visibility: Guarantees cloud-upload of stage reports even in partial failure scenarios.
     """
 
     run_context = RunContext.create()
@@ -223,7 +232,7 @@ def main() -> None:
     try:
 
         download_raw_snapshot(run_context)
-        initiliaze_metadata(run_context)
+        initialize_metadata(run_context)
 
         run_initial_validation_stage(run_context)
         run_contract_application_stage(run_context)

data_pipeline/semantic/registy.py data_pipeline/semantic/registry.pydata_pipeline/semantic/registy.py renamed to data_pipeline/semantic/registry.py

@@ -8,7 +8,7 @@
 from data_pipeline.shared.run_context import RunContext
 from data_pipeline.shared.loader_exporter import load_historical_table, export_file
 from data_pipeline.semantic.semantic_logic import init_report, log_error, log_info
-from data_pipeline.semantic.registy import SEMANTIC_MODULES
+from data_pipeline.semantic.registry import SEMANTIC_MODULES
 
 
 def task_wrapper(
@@ -111,11 +111,11 @@ def orchestrate_module(
     Coordinates the construction, validation, and export of a semantic module.
 
     Workflow:
-    1. Build: Executes the module-specific builder logic.
-    2. Loop: Iterates through each returned table in the builder output.
-    3. Validate: Enforces technical contracts (grain, schema, dtypes).
-    4. Export: Persists validated artifacts to the semantic zone.
-    5. Cleanup: Manages memory via explicit deletion and garbage collection.
+        1. Build: Executes the module-specific builder logic.
+        2. Loop: Iterates through each returned table in the builder output.
+        3. Validate: Enforces technical contracts (grain, schema, dtypes).
+        4. Export: Persists validated artifacts to the semantic zone.
+        5. Cleanup: Manages memory via explicit deletion and garbage collection.
 
     Invariants:
     - Fail-Fast: Any error in building or table-level processing halts the module.
@@ -188,10 +188,10 @@ def build_semantic_layer(run_context: RunContext) -> Dict:
     Main entry point for the Gold-to-Semantic stage.
 
     Workflow:
-    1. Source Verification: Loads 'assembled_events' and halts if empty/missing.
-    2. Registry Execution: Iterates through 'SEMANTIC_MODULES'.
-    3. Orchestration: Triggers builder logic followed by contract enforcement.
-    4. Cleanup: Purges memory after each module export.
+        1. Source Verification: Loads 'assembled_events' and halts if empty/missing.
+        2. Registry Execution: Iterates through 'SEMANTIC_MODULES'.
+        3. Orchestration: Triggers builder logic followed by contract enforcement.
+        4. Cleanup: Purges memory after each module export.
 
     Guarantees:
     - Atomicity: Module failures are trapped but mark the entire stage as 'failed'.

data_pipeline/semantic/semantic_executor.py

@@ -28,10 +28,10 @@ def apply_validation(run_context: RunContext, base_path: Path | None = None) ->
     subsequent Contract and Assembly stages.
 
     Workflow:
-    1. Loading: Iteratively fetches logical tables from the snapshot zone.
-    2. Base Check: Enforces schema, uniqueness, and null constraints via 'run_base_validations'.
-    3. Role Dispatch: Executes specialized logic (Event/Transaction) based on 'TABLE_CONFIG'.
-    4. Referential Check: Evaluates inter-table integrity (orphans) via 'run_cross_table_validations'.
+        1. Loading: Iteratively fetches logical tables from the snapshot zone.
+        2. Base Check: Enforces schema, uniqueness, and null constraints via 'run_base_validations'.
+        3. Role Dispatch: Executes specialized logic (Event/Transaction) based on 'TABLE_CONFIG'.
+        4. Referential Check: Evaluates inter-table integrity (orphans) via 'run_cross_table_validations'.
 
     Operational Guarantees:
     - Diagnostic Only: This function is read-only and will never mutate the source data.

data_pipeline/validation/validation_executor.py

@@ -0,0 +1,66 @@
+# **Assembly Stage**
+
+**Files:**
+* **Executor:** [`assembly_executor.py`](../data_pipeline/assembly/assembly_executor.py)
+* **Logic:** [`assembly_logic.py`](../data_pipeline/assembly/assembly_logic.py)
+
+**Role:** Data Integration and Analytical Flattening.
+
+## **System Contract**
+
+**Purpose**
+
+Integrates multiple normalized relational tables into a unified, analytical "Event" dataset and extracts high-fidelity "Dimension" references. It transforms raw business facts into a ready-to-model state by enforcing cardinality rules and calculating temporal performance metrics.
+
+**Invariants**
+* **Strict Order-ID Grain:** The primary event output is guaranteed to be exactly 1 row per `order_id`. Any operation causing cardinality explosion triggers a terminal failure.
+* **Inner-Join Priority:** To maintain analytical integrity, orders without corresponding items are purged.
+* **Temporal Determinism:** All lead times, lags, and delays are calculated as integer-day durations based on validated UTC timestamps.
+* **Reference Uniqueness:** Dimension reference tables (Customers, Products) are strictly deduplicated by their primary keys.
+
+**Inputs**
+* `run_context`: `RunContext` (Path resolution for Silver/contracted and Gold/assembled zones).
+* **Source Tables:** `df_orders`, `df_order_items`, `df_payments` (from the contracted layer).
+
+**Outputs**
+* **Assembly Report:** `dict` (Step-level status and informational logs).
+* **Assembled Events:** `parquet` (The unified analytical order-grain table).
+* **Dimension Refs:** `parquet` (Unique snapshots of customer and product attributes).
+
+## **Execution Workflow**
+
+The **Executor** coordinates two distinct sub-orchestrations:
+
+### **Workflow I: Event Assembly**
+1.  **Batch Load:** Fetches the required triplet (`orders`, `items`, `payments`) from the Silver zone.
+2.  **Merge:** Joins datasets using `merge_data`. It performs an inner join on items and a left join on payments to preserve financial data without losing order context.
+3.  **Derivation:** Executes `derive_fields` to calculate fulfillment lead times and extract ISO-calendar attributes.
+4.  **Schema Freeze:** Projects the final `ASSEMBLE_SCHEMA` and casts all columns to `ASSEMBLE_DTYPES`.
+5.  **Export & Clean:** Persists the table and triggers `gc.collect()` to free memory before dimension processing.
+
+### **Workflow II: Dimension Reference Extraction**
+1.  **Selection:** Iterates through the `DIMENSION_REFERENCES` registry.
+2.  **Deduplication:** Extracts the required column subset and drops duplicate primary keys.
+3.  **Export:** Persists each dimension (e.g., `df_customers`) as an independent artifact.
+
+## **Boundaries**
+
+| This component **DOES** | This component **DOES NOT** |
+| :--- | :--- |
+| Join multiple relational tables into a flat grain. | Perform data cleaning (handled in Contract stage). |
+| Calculate time-deltas (e.g., `lead_time_days`). | Perform complex multi-stage aggregations (delegated to Semantic stage). |
+| Enforce 1:1 cardinality for the final event grain. | Handle schema validation of raw data. |
+| Deduplicate dimension attributes. | Manage partitioning logic (managed by the loader/exporter). |
+| Manage peak memory via explicit `gc` triggers. | Change historical values or re-map IDs. |
+
+## **Failure & Severity Model**
+
+### **Operational Failures (System Level)**
+* **Loading Missing Table:** If a required table (e.g., `df_orders`) is missing from the Silver zone, the stage returns `failed` immediately.
+* **Export Failure:** Disk I/O errors or path resolution issues during the `export_file` call halt the lifecycle.
+
+### **Functional Findings (Data Level)**
+
+* **Cardinality Explosion:** If `merge_data` detects multiple rows for a single `order_id`, it raises a `RuntimeError`, treating it as a fatal violation of the analytical grain.
+* **Reference Duplication:** If a dimension table contains duplicate primary keys after extraction, it raises a `RuntimeError` to prevent downstream join corruption.
+* **Partial Payments:** Orders without payments are allowed (via Left Join); the system fills these with `None/NaN`, which is considered a valid business state rather than a failure.