DEV Community: Peter O’Mosagwe

Building a Production-Ready Daily Customer Orders Snapshot with Airflow, Python, and Oracle

Peter O’Mosagwe — Thu, 05 Feb 2026 05:22:27 +0000

Why This Matters (From Real Data Platforms)

In large data platforms, empty data is often more dangerous than bad data

A daily pipeline that “runs successfully” but ingests zero rows can quietly break downstream reports, fraud models, and executive dashboards. In one real-world scenario, a missing partition in an orders fact table propagated all the way to a revenue report — the pipeline was green, but business numbers were wrong for an entire day.

This is why mature data platforms treat data availability as a first-class validation, not an afterthought.

By combining:

Database-level guarantees (stored procedures)
Application-level orchestration (Python)
Operational enforcement (Airflow)

we build pipelines that fail loudly, alert early, and recover predictably.

Modern data pipelines often start with legacy stored procedures, but as data engineering teams scale, moving SQL logic into Python + Airflow pipelines provides better orchestration, observability, and validation.

In this article, we will walk through an industry-standard pipeline for a daily customer orders snapshot (Oracle-based) using:

Schemas: CO (customers, stores, products) and OE (orders, order_items)
Partitioned tables: Daily partitions for large fact tables
Validation checks: Fail if no rows exist, send emails
Airflow orchestration: Automated daily runs with success/failure notifications

showing how responsibilities are cleanly split across layers.

1️⃣ Sample Stored Procedure (Legacy Approach)

CREATE OR REPLACE PROCEDURE sp_daily_customer_orders_snapshot (
    p_partition_date IN VARCHAR2 DEFAULT NULL
) AS
    v_partition_date VARCHAR2(8);
BEGIN
    v_partition_date := COALESCE(p_partition_date, TO_CHAR(TRUNC(SYSDATE)-1, 'YYYYMMDD'));

    BEGIN
        EXECUTE IMMEDIATE 'DROP TABLE dw.daily_customer_orders';
    EXCEPTION
        WHEN OTHERS THEN
            IF SQLCODE != -942 THEN RAISE; END IF;
    END;

    EXECUTE IMMEDIATE '
        CREATE TABLE dw.daily_customer_orders AS
        SELECT /*+ PARALLEL(8) */
            c.customer_id,
            c.customer_name,
            s.store_id,
            s.store_name,
            o.order_id,
            o.order_date,
            oi.product_id,
            p.product_name,
            oi.quantity,
            oi.unit_price,
            (oi.quantity * oi.unit_price) AS total_amount,
            CASE
                WHEN (oi.quantity * oi.unit_price) >= 1000 THEN ''High''
                WHEN (oi.quantity * oi.unit_price) >= 500 THEN ''Medium''
                ELSE ''Low''
            END AS order_value_category
        FROM oe.orders_fct PARTITION (P' || v_partition_date || ') o
        INNER JOIN oe.order_items_fct PARTITION (P' || v_partition_date || ') oi
            ON o.order_id = oi.order_id
        INNER JOIN co.customers_dim c
            ON o.customer_id = c.customer_id
        INNER JOIN co.stores_dim s
            ON o.store_id = s.store_id
        INNER JOIN co.products_dim p
            ON oi.product_id = p.product_id
        WHERE o.order_status = ''COMPLETED''
    ';

    COMMIT;
END sp_daily_customer_orders_snapshot;
/

Assumption: The above procedure is called by Airflow.

How it’s typically run

Airflow (or Control-M / cron) calls the procedure
Logs go to DBMS_OUTPUT
If the table is empty, the procedure still “succeeds”

Limitations of this approach:

Hard to monitor & test
No automated alerts
No validation if the table is empty
Tightly coupled to Oracle

2️⃣ Refactored Python + Airflow Approach (Modern Approach)

The best pattern today is separation of concerns.

The golden rule

Oracle does heavy SQL. Python + Airflow do orchestration, validation, and alerting.

You do not rewrite SQL into pandas.

You move control, not computation.

Moving SQL to Python allows for dynamic orchestration, validation, and alerts.

So Python becomes the conductor, Oracle remains the engine.

Responsibilty split:

Layer	Responsibility
SQL files	Transformations (CTAS, joins, analytics)
Python	Parameters, validation, logging
Airflow	Scheduling, retries, emails, observability
Oracle	Execution engine

2.1 Externalize SQL (Keep it optimized)

SQL File: daily_customer_orders.sql

-- File: sql/daily_customer_orders.sql

CREATE TABLE dw.daily_customer_orders AS
SELECT /*+ PARALLEL(8) */
    c.customer_id,
    c.customer_name,
    s.store_id,
    s.store_name,
    o.order_id,
    o.order_date,
    oi.product_id,
    p.product_name,
    oi.quantity,
    oi.unit_price,
    (oi.quantity * oi.unit_price) AS total_amount,
    CASE
        WHEN (oi.quantity * oi.unit_price) >= 1000 THEN 'High'
        WHEN (oi.quantity * oi.unit_price) >= 500 THEN 'Medium'
        ELSE 'Low'
    END AS order_value_category
FROM oe.orders_fct PARTITION (P{{ partition_date }}) o
INNER JOIN oe.order_items_fct PARTITION (P{{ partition_date }}) oi
    ON o.order_id = oi.order_id
INNER JOIN co.customers_dim c
    ON o.customer_id = c.customer_id
INNER JOIN co.stores_dim s
    ON o.store_id = s.store_id
INNER JOIN co.products_dim p
    ON oi.product_id = p.product_id
WHERE o.order_status = 'COMPLETED';

Key idea:

SQL remains pure Oracle
Partition is injected dynamically > Placeholder {{ partition_date }} will be replaced dynamically in Python.

2.2 Python Execution Script

Design Principle
Oracle CTAs does not fail on empty results.
So we explicitly:

Run CTAS
Immediately SELECT COUNT(*)
If count = 0:
- Raise a custom exception
- Airflow handles alerting

Once any step returns zero rows -> exception raised -> task fails.

scripts/daily_customer_orders_pipeline.py

# File: scripts/daily_customer_orders_pipeline.py
import logging
from cx_Oracle import connect

log = logging.getLogger(__name__)

class EmptyTableError(Exception):
    """Raised when the target table is empty."""
    pass

def validate_rowcount(cursor, table_name: str):
    cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
    count = cursor.fetchone()[0]
    log.info("Row count for %s: %d", table_name, count)
    if count == 0:
        raise EmptyTableError(f"{table_name} has NO DATA!")

def run_daily_orders(conn, partition_date: str):
    cursor = conn.cursor()

    # Drop existing table
    try:
        cursor.execute("DROP TABLE dw.daily_customer_orders PURGE")
        log.info("Dropped existing table")
    except Exception as e:
        if "ORA-00942" in str(e):
            log.info("Table does not exist")
        else:
            raise

    # Read SQL and replace partition
    sql_file = "/opt/airflow/sql/daily_customer_orders.sql"
    with open(sql_file, "r") as f:
        sql = f.read().replace("{{ partition_date }}", partition_date)

    # Execute SQL
    cursor.execute(sql)

    # Validate row count
    validate_rowcount(cursor, "co_stg.daily_customer_orders")

    conn.commit()
    log.info("Snapshot created successfully for partition %s", partition_date)

Why this matters

CTAS with zero rows is treated as a failure
Bad data never silently propagates
Validation is reusable across pipelines

2.3 Airflow DAG with success & failure emails

dags/daily_customer_orders_dag.py

# File: dags/daily_customer_orders_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.hooks.oracle_hook import OracleHook
from datetime import datetime
from scripts.daily_customer_orders_pipeline import run_daily_orders, EmptyTableError

default_args = {
    "owner": "data-engineering",
    "email": ["data-alerts@company.com"],
    "email_on_failure": True,
    "email_on_success": True,
    "retries": 1,
    "retry_delay": 300,
}

def run_pipeline(**context):
    partition_date = context["ds_nodash"]  # YYYYMMDD
    hook = OracleHook(oracle_conn_id="oracle_default")
    conn = hook.get_conn()
    try:
        run_daily_orders(conn, partition_date)
    except EmptyTableError as e:
        raise RuntimeError(str(e))

with DAG(
    dag_id="daily_customer_orders_snapshot",
    default_args=default_args,
    schedule="@daily",
    start_date=datetime(2024, 1, 1),
    catchup=False,
    max_active_runs=1
) as dag:

    task_run_snapshot = PythonOperator(
        task_id="run_daily_customer_orders_snapshot",
        python_callable=run_pipeline,
        provide_context=True
    )

What happens at runtime

Success case
- Table created
- Row count > 0
- Task succeeds
- Success email sent
- Downstream tasks can run
Failure case (empty table)
- Validation fails
- Task errors intentionally
- Failure email sent
- DAG stops
- Issue visible immediately

This is data quality enforcement, not just scheduling.

3️⃣ Key Features

Partitioned Table Handling: Dynamic partition_date via Airflow execution date
Row-Count Validation: Task fails if no data → triggers email alert
Success/Failure Notifications: Automatic emails from Airflow
Idempotent: Drops old table before snapshot
Performance Optimized: Oracle PARALLEL hint for large datasets

Airflow will send something like:

Task Failed: run_customers_orders

Exception:

Step X - Customer Orders failed: Table has NO DATA

Execution date: 2026-02-03

Log URL:...

This is far more actionable than DBMS_OUTPUT.

4️⃣ Pros & Cons

Approach	Pros	Cons
Stored Procedure	Fast, data-local	Hard to monitor, test, alert
Python + Airflow	Observability, retries, email alerts, validation, partitioned pipeline	Requires setup & orchestration

5️⃣ Target Architecture

airflow/
├── dags/
│   └── daily_customer_orders_dag.py
├── scripts/
│   └── daily_customer_orders_pipeline.py
└── sql/
    ├── daily_customer_orders.sql

6️⃣ Takeaways

Stored procedures are not “bad”. They are just not orchestration tools. There is no universal winner between stored procedures and SQL-in-Python — strong pipelines use both, intentionally.

Move heavy SQL to Oracle, but orchestration, validation, and monitoring to Python + Airflow.
Implement row-count validation and alerts to catch empty or failed datasets early.
Modular architecture allows scaling: more tables, incremental loads, retries, and notifications.

Recommendations

You can refactor and enhance for different use cases:

Incremental Loads: Instead of full snapshot, insert only new records.
Branching DAGs: Skip downstream tasks if no data.

Repository Pattern in a Data Analytics / ML Pipeline

Peter O’Mosagwe — Mon, 02 Feb 2026 17:17:46 +0000

1. The Core Problem in Most ML Projects

Let’s start with something very real.
In many analytics and ML projects, data access looks like this:

df = pd.read_sql("""
    SELECT customer_id,
           SUM(amount) AS total_amount
    FROM transactions
    WHERE transaction_date >= '2025-01-01'
    GROUP BY customer_id
""", conn)

At first, this feels efficient.

But over time:

The SQL grows
Business logic sneaks into queries
Feature logic is duplicated
Nobody knows which query feeds which model
Testing becomes painful

Eventually, the ML pipeline becomes a tightly coupled mess.

👉 The Repository Pattern exists to stop this from happening.

2. The Big Idea (In Plain English)

“My ML pipeline should not know where the data comes from — only what data it needs.”

So instead of saying:

“Give me data from MySQL using this SQL query”

Your pipeline says:

“Give me all transactions between these dates”

That’s it.

3. Mental Model: Repositories as Data Translators

Think of a repository as a translator between:

🧠 Business / ML logic
and
🗄️ Physical data storage (MySQL)

ML Pipeline speaks:     "transactions"
Database speaks:       "tables, joins, SQL"
Repository speaks both.

The repository hides:

SQL
Connection handling
Database quirks
Performance tuning

4. ML Pipeline with Repository Pattern (Conceptual View)

MySQL Database
     |
     |  (SQL, connections, credentials)
     v
Repository Layer
     |
     |  (clean Python objects)
     v
Feature Engineering
     |
     v
Model Training

Important rule:
⬇️ Dependencies only go down, never up.

5. Step 1: Define What the Pipeline Cares About (Domain Model)

In analytics, we don’t need heavy ORM entities.

We just need meaningful data structures.

from dataclasses import dataclass
from datetime import date


@dataclass
class Transaction:
    customer_id: int
    amount: float
    transaction_date: date
    transaction_type: str

Why this matters

No SQL
No MySQL
No Pandas
Pure Python

This is domain language: something analysts and ML engineers understand.

6. Step 2: Define the Repository Contract (The Promise)

Now we ask:

What data does the ML pipeline need?

Not how to get it, just what.

from abc import ABC, abstractmethod
from typing import List
from datetime import date


class TransactionRepository(ABC):


    @abstractmethod
    def get_transactions(
        self,
        start_date: date,
        end_date: date
    ) -> List[Transaction]:
        pass

Key idea here

This is a promise:

Any data source
Any database
Any storage engine

…as long as it fulfills this contract.

7. Why This Is Powerful

At this point:

The ML pipeline depends on an interface
It does NOT depend on MySQL
It does NOT depend on PyMySQL

This is what gives you:

Testability
Flexibility
Clean design

8. Step 3: Implement the Repository Using PyMySQL

Now — and only now — do we touch MySQL.

Connection helper

import pymysql


def get_connection():
    return pymysql.connect(
        host="localhost",
        user="analytics_user",
        password="analytics_pwd",
        database="analytics_db",
        cursorclass=pymysql.cursors.DictCursor
    )

This stays out of the ML logic.

MySQL-backed Repository

class MySQLTransactionRepository(TransactionRepository):


    def get_transactions(self, start_date, end_date):


        query = """
            SELECT customer_id,
                   amount,
                   transaction_date,
                   transaction_type
            FROM transactions
            WHERE transaction_date BETWEEN %s AND %s
        """


        conn = get_connection()
        try:
            with conn.cursor() as cursor:
                cursor.execute(query, (start_date, end_date))
                rows = cursor.fetchall()


            return [
                Transaction(
                    customer_id=row["customer_id"],
                    amount=float(row["amount"]),
                    transaction_date=row["transaction_date"],
                    transaction_type=row["transaction_type"]
                )
                for row in rows
            ]
        finally:
            conn.close()

What just happened?

SQL is isolated
Connection lifecycle is controlled
Raw rows are converted into domain objects

Everything else in the system stays clean.

9. Step 4: Feature Engineering (Pure Data Logic)

This layer doesn’t know:

Where data came from
How it was queried
Whether it was MySQL, CSV, or API

import pandas as pd


class TransactionFeatureEngineer:


    def build_customer_features(self, transactions):


        df = pd.DataFrame([t.__dict__ for t in transactions])


        features = (
            df.groupby("customer_id")
              .agg(
                  total_amount=("amount", "sum"),
                  avg_amount=("amount", "mean"),
                  txn_count=("amount", "count")
              )
              .reset_index()
        )


        return features

Why this is clean

Deterministic
Easy to unit test
No side effects
Reusable across models

10. Step 5: Model Training Layer

Again — no database awareness.

from sklearn.ensemble import RandomForestClassifier


class ModelTrainingService:


    def train(self, X, y):
        model = RandomForestClassifier(
            n_estimators=100,
            random_state=42
        )
        model.fit(X, y)
        return model

The model only cares about features, not data sources.

11. Step 6: Orchestrating the Pipeline

This is the only place everything comes together.

from datetime import date


repo = MySQLTransactionRepository()
feature_engineer = TransactionFeatureEngineer()
trainer = ModelTrainingService()


# Fetch
transactions = repo.get_transactions(
    date(2025, 1, 1),
    date(2025, 12, 31)
)


# Features
features_df = feature_engineer.build_customer_features(transactions)


# Example target
features_df["target"] = (
    features_df["total_amount"] > 100000
).astype(int)


X = features_df[["total_amount", "avg_amount", "txn_count"]]
y = features_df["target"]


# Train
model = trainer.train(X, y)

This code reads like a story, not plumbing.

12. The Real Superpower: Testing Without MySQL

Now comes the magic.

In-memory repository

class InMemoryTransactionRepository(TransactionRepository):


    def __init__(self, transactions):
        self.transactions = transactions


    def get_transactions(self, start_date, end_date):
        return [
            t for t in self.transactions
            if start_date <= t.transaction_date <= end_date
        ]

Why this is huge

No database
No credentials
No network
Instant tests

This is how serious ML platforms are built.

13. Common ML Anti-Patterns This Avoids

Anti-pattern Why it’s bad
SQL in notebooks Hard to version & test
Feature logic in SQL Locks logic to DB
DB calls in training Non-deterministic
Hard-coded tables Brittle pipelines

The repository pattern fixes all of them.

14. When This Pattern Really Shines

Especially useful for:

Fraud detection
Credit scoring
Customer segmentation
Regulatory analytics
Feature reuse across models

(Yes — this is very bank-friendly 😄)

15. Final Intuition to Remember

Repositories answer “WHAT data?”
Databases answer “HOW stored?”
ML answers “WHAT to learn?”

Keep those questions separate — and your pipelines stay sane.