docs: refactor docstrings to procedural and tight style

Author: BLMgithub

data_pipeline/run_pipeline.py

@@ -103,25 +103,31 @@ def finalize_run(run_context: RunContext, status: str) -> None:
 
 def main() -> None:
     """
-    Pipeline execution controller.
+    Pipeline orchestrator.
 
     Execution order:
-
-    1. Initialize run context and directory structure.
-    2. Capture raw snapshot and initialize metadata.
-    3. Run initial validation on raw data.
-       - Exit if structural errors exist.
-    4. Apply table contracts in configured parent → child order,
-       propagating invalid order_ids.
-    5. Rerun validation on contracted data.
-       - Exit if any errors or warnings remain.
-    6. Assemble the core event table.
-       - Exit on assembly failure.
-    7. Build semantic layer tables.
-       - Exit on semantic failure.
-    8. Run pre-publish semantic integrity gate.
-       - Exit if gate fails.
-    9. Exit process with success code.
+    1. Snapshot raw data
+    2. Initialize metadata (RUNNING)
+    3. Validation → halt on errors
+    4. Contract enforcement
+        - Apply role-driven repair
+        - Propagate invalid order_ids (parent → child)
+    5. Re-validation → halt on errors/warnings
+    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
     """
 
     run_context = RunContext.create()

data_pipeline/shared/run_context.py

@@ -17,6 +17,20 @@ def _generate_run_id() -> str:
 
 @dataclass
 class RunContext:
+    """
+    Run-scoped execution context.
+
+    Responsibilities:
+    - Generate and hold run_id
+    - Define all stage directory paths
+    - Provide consistent path resolution across stages
+    - Enforce run isolation via run-scoped folders
+
+    Guarantees:
+    - Each run writes only within its own directory tree
+    - Published artifacts are resolved deterministically from run_id
+    """
+
     run_id: str
     base_path: str | Path
 

data_pipeline/stages/apply_raw_data_contract.py

@@ -136,26 +136,28 @@ def apply_contract(
     run_context: RunContext, table_name: str, invalid_order_ids: set | None = None
 ) -> tuple[dict, set]:
     """
-    Applies role-driven deterministic normalization, producing a cleaned
-    dataset and a structured execution report.
-
-    Chronological behavior:
-    - Initializes contract metrics and error container.
-    - Validates table eligibility against TABLE_CONFIG.
-    - Loads the logical table from the raw snapshot.
-    - Applies role-specific contract steps:
-        - **event_fact:**
-            - exact deduplication
-            - timestamp parse enforcement
-            - temporal invariant enforcement (produces invalid `order_ids`)
-        - **transaction_detail:**
-            - exact deduplication
-            - optional cascade removal using upstream invalid `order_ids`
-        - **entity_reference:**
-            - exact deduplication only.
-    - Records row-level impact for each enforcement step.
-    - Writes the contracted output to the contract layer.
-    - Returns the execution report and any newly invalidated `order_ids`.
+    Enforce structural contract on a single logical table.
+
+    Role-driven behavior:
+    - event_fact:
+        - exact deduplication
+        - remove unparsable timestamps
+        - remove temporal violations
+        - emit invalid order_ids
+    - transaction_detail:
+        - deduplicate
+        - cascade drop invalid order_ids
+    - entity_reference:
+        - deduplicate only
+
+    Guarantees:
+    - Deterministic row removal
+    - No numeric or domain correction
+    - Output written to contracted layer
+
+    Returns:
+    - Contract execution report
+    - Newly invalidated order_ids (if any)
     """
 
     report = {

data_pipeline/stages/assemble_validated_events.py

@@ -10,7 +10,7 @@
 import pandas as pd
 from typing import Dict, List
 from data_pipeline.shared.run_context import RunContext
-from data_pipeline.shared.table_configs import (
+from data_pipeline.shared.modeling_configs import (
     ASSEMBLE_ENFORCED_SCHEMA,
     ASSEMBLE_ENFORCED_DTYPES,
 )
@@ -158,19 +158,22 @@ def freeze_schema(df: pd.DataFrame) -> pd.DataFrame:
 
 def assemble_events(run_context: RunContext) -> Dict:
     """
-    Produces the contract-compliant event fact table from validated logical inputs and produce a consolidated run report
-
-    Chronological behavior:
-
-    - Initializes run-scoped reporting and logging helpers.
-    - Loads all required event-source tables from the contracted layer.
-    - Fails fast if any required table is missing or empty.
-    - Executes core assembly pipeline:
-      - merge_data (grain enforcement)
-      - derive_fields (event enrichment)
-      - freeze_schema (final contract projection)
-    - Exports the assembled dataset to the run-scoped output path.
-    - Aggregates all findings into the returned report.
+    Assemble contract-compliant event dataset (order grain).
+
+    Steps:
+    - Load contracted event tables
+    - Merge with cardinality enforcement (1 row per order_id)
+    - Derive temporal metrics and lineage fields
+    - Freeze schema and enforce dtypes
+    - Export deterministic output
+
+    Grain:
+    - One row per order_id (hard fail on violation)
+
+    Non-responsibilities:
+    - No validation
+    - No repair
+    - No business logic
     """
 
     report = init_report()

data_pipeline/stages/build_bi_semantic_layer.py

@@ -8,7 +8,7 @@
 import pandas as pd
 from typing import Dict, List, Tuple, Literal
 from data_pipeline.shared.run_context import RunContext
-from data_pipeline.shared.table_configs import (
+from data_pipeline.shared.modeling_configs import (
     SELLER_DIM_ENFORCED_SCHEMA,
     SELLER_DIM_ENFORCED_DTYPES,
     SELLER_FACT_ENFORCED_SCHEMA,
@@ -45,23 +45,22 @@ def seller_weekly_semantic(
     df: pd.DataFrame,
 ) -> Tuple[pd.DataFrame, pd.DataFrame]:
     """
-    Seller weekly semantic builder.
-
-    Transforms the assembled event table into seller-level weekly
-    performance fact and supporting seller dimension.
-
-    Transform behavior:
-
-    - Validates single-run lineage via `run_id`
-    - Derives weekly alignment fields and status flags
-    - Aggregates event data to seller-week grain
-    - Builds seller dimension from first observed activity
+    Build seller weekly semantic layer from assembled events.
 
     Fact grain:
-    - One row per (seller_id, order_year_week)
+    - 1 row per (seller_id, order_year_week)
 
     Dimension grain:
-    - One row per seller_id
+    - 1 row per seller_id
+
+    Behavior:
+    - Enforce single run_id lineage
+    - Derive ISO week alignment
+    - Aggregate event metrics to seller-week
+
+    Returns:
+    - Aggregated fact dataframe
+    - Seller dimension dataframe
     """
 
     read_assembled = df.copy()
@@ -109,40 +108,26 @@ def freeze_seller_semantic(
     table_type: Literal["fact", "dim"],
 ) -> pd.DataFrame:
     """
-    Seller semantic contract enforcer.
+    Enforce seller semantic contract.
 
-    Routes the input table to the appropriate fact or dimension
-    contract freezer and enforces grain integrity before projection.
+    For fact:
+    - Validate required columns
+    - Enforce (seller_id, order_year_week) uniqueness
+    - Apply projection, dtype enforcement, deterministic sort
 
-    Behavior:
+    For dimension:
+    - Validate required columns
+    - Enforce seller_id uniqueness
+    - Apply projection and dtype enforcement
 
-    - Validates `table_type` selector
-    - Applies grain-level duplicate checks:
-      - fact: (seller_id, order_year_week)
-      - dim: seller_id
-    - Dispatches to the corresponding schema freezer
-    - Returns a BI-ready, schema-stable dataframe
+    Raise:
+    - RuntimeError on schema or grain violation
     """
 
     if table_type not in {"fact", "dim"}:
         raise ValueError
 
     def freeze_seller_fact(df: pd.DataFrame) -> pd.DataFrame:
-        """
-        Seller weekly fact contract enforcement.
-
-        Projects the aggregated seller-week dataset into the approved
-        fact schema, enforces dtypes, and applies deterministic ordering.
-
-        Enforcement actions:
-
-        - Validates presence of all required fact columns
-        - Projects to the contract column order
-        - Casts fields to enforced dtypes
-        - Sorts by (seller_id, order_year_week)
-        - Resets index for clean downstream consumption
-
-        """
 
         fact_contract = df.copy()
 
@@ -161,20 +146,6 @@ def freeze_seller_fact(df: pd.DataFrame) -> pd.DataFrame:
         return fact_contract
 
     def freeze_seller_dim(df: pd.DataFrame) -> pd.DataFrame:
-        """
-        Seller dimension contract enforcement.
-
-        Projects the seller dimension into the approved schema,
-        enforces dtypes, and applies deterministic ordering.
-
-        Enforcement actions:
-
-        - Validates presence of all required dimension columns
-        - Projects to the contract column order
-        - Casts fields to enforced dtypes
-        - Sorts by seller_id
-        - Resets index for clean downstream consumption
-        """
 
         dim_contract = df.copy()
 
@@ -216,20 +187,24 @@ def build_semantic_layer(run_context: RunContext) -> Dict:
     """
     Semantic layer orchestrator.
 
-    Builds seller performance semantic tables from the assembled
-    event layer and exports contract-compliant BI artifacts.
+    Builds semantic modules from the assembled event layer and
+    exports contract-compliant BI artifacts.
+
+    Execution:
+    - Load assembled event dataset (order grain)
+    - Fail fast if dataset is missing or empty
+    - Execute registered semantic builders
+    - Apply module-level freeze contracts
+    - Export semantic artifacts into run-scoped semantic directory
+    - Aggregate findings into report
 
-    Chronological behavior:
+    Guarantees:
+    - Each semantic module enforces its declared grain
+    - All exported tables are contract-compliant
+    - No decision logic embedded
 
-    - Initializes run-scoped reporting and logging helpers.
-    - Loads the assembled_events logical table.
-    - Fails fast if the assembled dataset is missing or empty.
-    - Executes semantic pipeline:
-      - seller_weekly_semantic (aggregation)
-      - freeze_seller_semantic (fact and dimension contracts)
-    - Generates run-partitioned output filenames.
-    - Exports semantic tables to the run-scoped semantic directory.
-    - Aggregates all findings into the returned report.
+    Failure:
+    - Any module failure halts semantic stage
     """
 
     report = init_report()