AI Data Engineer快速实战班

# RetailFlow local reference

Synthetic, pre-generated teaching reference. Run `python3 pipeline.py --output result.json` using Python 3 with its standard library. No installation, network, API key, AI call, cloud account or charge is needed.

`orders-batch-1.csv` and `lines-batch-1.csv` produce AUD 150 revenue, 2 orders, 4 units and AUD 75 average order value after duplicate line removal. Batch 2 cancels O101 and adds O102, producing AUD 130, 2 orders, 4 units and AUD 65 average order value. Repeating batch 2 leaves the result unchanged. An older O101 paid status cannot override the cancellation. A deliberately injected P999 line must fail validation; it is rolled back and the last valid result is retained.

The SQL models and Python runner are the complete implementation of this small SQLite reference. The course's planned Snowflake/dbt models, cloud permissions, scheduler and AI workflow are not implemented here. This is not a production platform or a finished implementation of every course outcome.

Scope: a single predefined business date, fixed AUD, current-value product categories, immutable line keys. Changed existing line amounts, equal-timestamp conflicts, full source reconciliation, tax/refund accounting, multiple dates, concurrency and production access controls are outside this reference. Explain these limitations when presenting the project.

Ownership: course reference template. A learner must add and verify their own change before describing personal contributions. Do not present the reference as real client work, real learner work, or measured career outcomes.

"""RetailFlow local reference: stdlib SQLite, synthetic data, no cloud/API calls.
Run: python3 pipeline.py --output result.json
This validates the teaching logic, not the planned Snowflake/dbt implementation.
"""
import argparse
import csv
import json
import sqlite3
from pathlib import Path

ROOT = Path(__file__).resolve().parent


def execute():
    db = sqlite3.connect(':memory:')
    db.row_factory = sqlite3.Row
    db.executescript((ROOT / 'models.sql').read_text())

    def load_orders(filename):
        for row in csv.DictReader((ROOT / filename).open()):
            db.execute('''INSERT INTO orders VALUES (:order_id,:status,:updated_at)
                ON CONFLICT(order_id) DO UPDATE SET status=excluded.status,
                updated_at=excluded.updated_at WHERE excluded.updated_at > orders.updated_at''', row)

    def load_lines(filename):
        for row in csv.DictReader((ROOT / filename).open()):
            db.execute('''INSERT INTO lines VALUES (:order_id,:line_id,:product_id,:quantity,:unit_price)
                ON CONFLICT(order_id,line_id) DO NOTHING''', row)

    def validate():
        missing = db.execute('''SELECT l.order_id,l.line_id,l.product_id FROM lines l
            LEFT JOIN products p USING(product_id) WHERE p.product_id IS NULL''').fetchall()
        if missing:
            raise ValueError('Missing product reference: ' + json.dumps([dict(r) for r in missing]))
        invalid = db.execute('SELECT COUNT(*) FROM lines WHERE quantity<=0 OR unit_price<0').fetchone()[0]
        if invalid:
            raise ValueError('Invalid quantity or unit price')

    def snapshot():
        validate()
        row = dict(db.execute('SELECT * FROM daily_sales').fetchone())
        row['order_count'] = int(row['order_count'])
        return row

    db.executemany('INSERT INTO products VALUES (?,?)', [('P100','Accessories'),('P101','Stationery'),('P102','Home')])
    load_orders('orders-batch-1.csv')
    load_lines('lines-batch-1.csv')
    first = snapshot()
    assert first['revenue'] == 150 and first['order_count'] == 2 and first['units'] == 4
    load_orders('orders-batch-2.csv')
    load_lines('lines-batch-2.csv')
    second = snapshot()
    assert second['revenue'] == 130 and second['aov'] == 65
    load_orders('orders-batch-2.csv')
    load_lines('lines-batch-2.csv')
    repeated = snapshot()
    assert second == repeated
    load_orders('orders-late.csv')
    assert snapshot() == second, 'Older paid status must not overwrite latest cancellation'
    db.execute('SAVEPOINT invalid_batch')
    db.execute("INSERT INTO lines VALUES ('O102','L2','P999',1,99)")
    try:
        snapshot()
    except ValueError as error:
        failure = str(error)
    else:
        raise AssertionError('Missing product should block result publication')
    db.execute('ROLLBACK TO invalid_batch')
    restored = snapshot()
    assert restored == second
    return {
        'evidence': 'Executed local SQLite reference; synthetic data; no Snowflake/dbt or scheduler validation',
        'business_date': '2026-10-01',
        'currency': 'AUD',
        'stages': [
            {'id':'first','label':'第一批：完成去重','status':'passed','result':first,
             'explanation':'重复的 O100-L1 只保留一次，三条有效明细对应两个订单。'},
            {'id':'second','label':'第二批：更新与取消','status':'passed','result':second,
             'explanation':'O101 最新状态变为 cancelled，移出 A$50；O102 新增 A$30。'},
            {'id':'repeat','label':'同批重跑','status':'passed','result':repeated,
             'explanation':'第二批再次执行，业务结果与上次完全一致；更旧的 paid 状态也不会覆盖取消。'},
            {'id':'failure','label':'异常批次：阻止发布','status':'blocked','result':restored,
             'explanation':'P999 不存在，检查失败，异常批次回滚；这里显示上一次有效结果，不是本次新结果。','error':failure},
        ],
        'checks': [
            {'name':'Duplicate line removed','status':'passed'},
            {'name':'Cancellation applied','status':'passed'},
            {'name':'Repeated batch unchanged','status':'passed'},
            {'name':'Older status ignored','status':'passed'},
            {'name':'Missing product blocks publication','status':'passed'},
            {'name':'Rollback retains last valid result','status':'passed'},
        ],
        'limitations': ['One business date; fixed AUD; no tax/refund accounting',
                         'Line changes with the same key are outside this reference implementation',
                         'Equal update timestamps require a defined tie-breaker in the course implementation',
                         'No cloud, AI execution, scheduling, or production deployment demonstrated'],
    }


if __name__ == '__main__':
    parser = argparse.ArgumentParser()
    parser.add_argument('--output', type=Path, default=ROOT / 'result.json')
    args = parser.parse_args()
    args.output.write_text(json.dumps(execute(), ensure_ascii=False, indent=2) + '\n')
    print('Local reference executed: 6 checks passed; wrote', args.output)

CREATE TABLE products(product_id TEXT PRIMARY KEY, category TEXT NOT NULL);
CREATE TABLE orders(order_id TEXT PRIMARY KEY, status TEXT NOT NULL, updated_at TEXT NOT NULL);
CREATE TABLE lines(
    order_id TEXT NOT NULL REFERENCES orders(order_id), line_id TEXT NOT NULL,
    product_id TEXT NOT NULL, quantity INTEGER NOT NULL, unit_price REAL NOT NULL,
    PRIMARY KEY(order_id,line_id)
);
-- One-date reference. All synthetic orders belong to 2026-10-01 Brisbane time.
-- The course Snowflake/dbt implementation must derive the business date explicitly.
CREATE VIEW valid_lines AS
SELECT l.*, p.category FROM lines l JOIN orders o USING(order_id)
JOIN products p USING(product_id) WHERE o.status IN ('paid','completed');
CREATE VIEW daily_sales AS
SELECT ROUND(COALESCE(SUM(quantity*unit_price),0),2) AS revenue,
       COUNT(DISTINCT order_id) AS order_count,
       COALESCE(SUM(quantity),0) AS units,
       ROUND(SUM(quantity*unit_price)/NULLIF(COUNT(DISTINCT order_id),0),2) AS aov
FROM valid_lines;

order_id,status,updated_at
O100,paid,2026-10-01T00:00:00Z
O101,paid,2026-10-01T00:05:00Z

order_id,line_id,product_id,quantity,unit_price
O100,L1,P100,2,40
O100,L1,P100,2,40
O100,L2,P101,1,20
O101,L1,P102,1,50

order_id,status,updated_at
O101,cancelled,2026-10-02T00:00:00Z
O102,paid,2026-10-02T00:01:00Z

order_id,line_id,product_id,quantity,unit_price
O102,L1,P100,1,30

order_id,status,updated_at
O101,paid,2026-10-01T00:05:00Z

{
  "evidence": "Executed local SQLite reference; synthetic data; no Snowflake/dbt or scheduler validation",
  "business_date": "2026-10-01",
  "currency": "AUD",
  "stages": [
    {
      "id": "first",
      "label": "第一批：完成去重",
      "status": "passed",
      "result": {
        "revenue": 150.0,
        "order_count": 2,
        "units": 4,
        "aov": 75.0
      },
      "explanation": "重复的 O100-L1 只保留一次，三条有效明细对应两个订单。"
    },
    {
      "id": "second",
      "label": "第二批：更新与取消",
      "status": "passed",
      "result": {
        "revenue": 130.0,
        "order_count": 2,
        "units": 4,
        "aov": 65.0
      },
      "explanation": "O101 最新状态变为 cancelled，移出 A$50；O102 新增 A$30。"
    },
    {
      "id": "repeat",
      "label": "同批重跑",
      "status": "passed",
      "result": {
        "revenue": 130.0,
        "order_count": 2,
        "units": 4,
        "aov": 65.0
      },
      "explanation": "第二批再次执行，业务结果与上次完全一致；更旧的 paid 状态也不会覆盖取消。"
    },
    {
      "id": "failure",
      "label": "异常批次：阻止发布",
      "status": "blocked",
      "result": {
        "revenue": 130.0,
        "order_count": 2,
        "units": 4,
        "aov": 65.0
      },
      "explanation": "P999 不存在，检查失败，异常批次回滚；这里显示上一次有效结果，不是本次新结果。",
      "error": "Missing product reference: [{\"order_id\": \"O102\", \"line_id\": \"L2\", \"product_id\": \"P999\"}]"
    }
  ],
  "checks": [
    {
      "name": "Duplicate line removed",
      "status": "passed"
    },
    {
      "name": "Cancellation applied",
      "status": "passed"
    },
    {
      "name": "Repeated batch unchanged",
      "status": "passed"
    },
    {
      "name": "Older status ignored",
      "status": "passed"
    },
    {
      "name": "Missing product blocks publication",
      "status": "passed"
    },
    {
      "name": "Rollback retains last valid result",
      "status": "passed"
    }
  ],
  "limitations": [
    "One business date; fixed AUD; no tax/refund accounting",
    "Line changes with the same key are outside this reference implementation",
    "Equal update timestamps require a defined tie-breaker in the course implementation",
    "No cloud, AI execution, scheduling, or production deployment demonstrated"
  ]
}