10retry ช่วยได้กับความพังแบบไหน

# ~/airflow-lab/dags/stock_monthly.py — บทที่ 10: ลองใหม่ได้ 2 ครั้ง ยกเว้นไฟล์ผิดรูปที่หยุดทันที
from datetime import datetime, timedelta
from pathlib import Path

import stock_gate
import stock_steps as steps
from airflow.sdk import dag, task
from airflow.sdk.exceptions import AirflowFailException, AirflowSkipException

LAB = Path.home() / "airflow-lab"


@dag(
    schedule=None,
    start_date=datetime(2026, 5, 1),
    catchup=False,
    params={"month": "2026-07"},
    default_args={"retries": 2, "retry_delay": timedelta(seconds=20)},  # แล็บรอ 20 วินาที ของจริงหลายนาที
)
def stock_monthly():
    @task
    def arrived(**context):
        month = context["params"]["month"]
        files = sorted((LAB / "inbox").glob(f"stock_{month}_*.xlsx"))
        if not files:  # ยังไม่มีไฟล์ของเดือนนี้สักสาขา — ไม่มีอะไรให้ทำ ไม่ใช่ความผิดของใคร
            raise AirflowSkipException(f"ยังไม่มีไฟล์ของเดือน {month}")
        return len(files)

    @task
    def receive(branch, **context):
        month = context["params"]["month"]
        path = LAB / "inbox" / f"stock_{month}_{branch}.xlsx"
        if not path.exists():  # ไฟล์มาช้า — ชั่วคราว ปล่อยให้ retries ลองใหม่
            raise FileNotFoundError(f"ไฟล์สาขา {stock_gate.NAMES[branch]} เดือน {month} ยังมาไม่ถึง")
        rows, missing = steps.read_branch(path)
        staged = steps.stage(branch, rows, missing, LAB / "staging" / month / f"{branch}.json")
        return {**staged, "problems": stock_gate.inspect(path, rows, missing, month)}

    @task
    def gate(staged):
        for s in staged:
            print(f"ด่าน ← receive_{s['branch']}: {s['rows']} แถว · ปัญหา {len(s['problems'])} ข้อ")
        try:
            return stock_gate.check(staged)
        except stock_gate.BadFile as e:  # ไฟล์ผิดรูป ลองกี่ครั้งก็ไม่หาย — หยุดเลย ไม่ใช้ retries ที่เหลือ
            raise AirflowFailException(str(e)) from e

    @task
    def transform(staged, **context):
        return steps.transform(staged, LAB / "staging" / context["params"]["month"] / "all.json")

    @task
    def load(summary, **context):
        return steps.load(summary, LAB / "warehouse.db", context["params"]["month"])

    @task
    def notify(staged, summary, in_db):
        print(steps.report(staged, summary, in_db))

    received = [receive.override(task_id=f"receive_{b}")(b) for b in "abc"]
    arrived() >> received
    staged = gate(received)
    summary = transform(staged)
    notify(staged, summary, load(summary))


stock_monthly()

airflow dags test stock_monthly -c '{"month": "2026-09"}' 2>&1 | grep "^FileNotFoundError"

FileNotFoundError: ไฟล์สาขา ค เดือน 2026-09 ยังมาไม่ถึง
FileNotFoundError: ไฟล์สาขา ค เดือน 2026-09 ยังมาไม่ถึง
FileNotFoundError: ไฟล์สาขา ค เดือน 2026-09 ยังมาไม่ถึง

python ~/airflow-lab/states.py stock_monthly

รอบล่าสุดของ stock_monthly {"month": "2026-09"} → failed
  arrived       success 1
  receive_a     success 1
  receive_b     success 1
  receive_c     failed 3
  gate          upstream_failed 0
  transform     upstream_failed 0
  load          upstream_failed 0
  notify        upstream_failed 0

── หน้าต่างที่ 1 ──
$ airflow dags test stock_monthly -c '{"month": "2026-09"}' 2>&1 | grep -E "^FileNotFoundError|AirflowFailException:"
FileNotFoundError: ไฟล์สาขา ค เดือน 2026-09 ยังมาไม่ถึง
airflow.sdk.exceptions.AirflowFailException: ด่านหยุดท่อ — สาขา ก: ไม่มีคอลัมน์ จำนวน

── หน้าต่างที่ 2 ระหว่างที่ receive_c รอรอบสอง ──
$ mv ~/airflow-lab/inbox/late/stock_2026-09_c.xlsx ~/airflow-lab/inbox/

── หลังรอบจบ ──
$ python ~/airflow-lab/states.py stock_monthly
รอบล่าสุดของ stock_monthly {"month": "2026-09"} → failed
  arrived       success 1
  receive_a     success 1
  receive_b     success 1
  receive_c     success 2
  gate          failed 1
  transform     upstream_failed 0
  load          upstream_failed 0
  notify        upstream_failed 0