실습 1 — 작은 워크플로 실행기 만들기

from graphlib import CycleError, TopologicalSorter

# 밤마다 도서관 안내 문서로 색인을 만드는 작업들: {작업: [먼저 끝나야 하는 작업]}
graph = {"문서 읽기": [], "검사": ["문서 읽기"], "조각 나누기": ["검사"],
         "임베딩": ["조각 나누기"], "교체": ["임베딩"], "보고": ["조각 나누기"]}
# static_order(): 앞 작업이 늘 먼저 오도록 줄 세운다
print("차례:", list(TopologicalSorter(graph).static_order()))

# 실수로 문서 읽기가 교체 뒤에 오게 적으면 되돌아오는 길(순환)이 생긴다
bad = dict(graph, **{"문서 읽기": ["교체"]})
try:
    list(TopologicalSorter(bad).static_order())
except CycleError as e:
    # 결과: 순환을 이루는 작업들
    print("순환:", e.args[1])

from pathlib import Path

# 실습 3, 6과 프로젝트가 불러 쓸 실행기를 파일로 만든다
Path("out_flow.py").write_text('''"""작은 워크플로 실행기(19장). Airflow 의 DAG 실행을 흉내 낸다."""
import time
from graphlib import TopologicalSorter


class Stop(Exception):
    """멈춤: 고장이 아니라 '더 갈 필요가 없다'는 신호. 다시 시도하지 않고 뒤 작업을 건너뛴다."""


def run_flow(tasks, retries=1, wait=0.0, log=print):
    """tasks: {이름: (함수, [앞 작업 이름, ...])}. 함수는 ctx(앞 작업들이 돌려준 값을 이름으로 담은 사전)를 받는다.
    돌려주는 값: (작업마다의 기록 목록, ctx)"""
    order = list(TopologicalSorter({n: deps for n, (f, deps) in tasks.items()}).static_order())
    ctx, state, records = {}, {}, []
    for name in order:
        func, deps = tasks[name]
        if any(state[d] != "성공" for d in deps):            # 앞 작업이 하나라도 성공하지 않았으면 건너뛴다
            state[name] = "건너뜀"
            records.append({"작업": name, "상태": "건너뜀", "시도": 0, "초": 0.0, "메모": ""})
            log(f"[건너뜀] {name}")
            continue
        for attempt in range(1, retries + 2):                # 처음 한 번 + 다시 시도 retries 번
            start = time.perf_counter()
            try:
                ctx[name] = func(ctx)
                state[name], memo = "성공", ""
            except Stop as e:
                state[name], memo = "멈춤", str(e)
            except Exception as e:
                state[name], memo = "실패", f"{type(e).__name__}: {e}"
            sec = round(time.perf_counter() - start, 2)
            log(f"[{state[name]}] {name} ({attempt}번째 시도)" + (f" · {memo}" if memo else ""))
            if state[name] != "실패" or attempt == retries + 1:
                break
            time.sleep(wait)
        records.append({"작업": name, "상태": state[name], "시도": attempt, "초": sec, "메모": memo})
    return records, ctx
''', encoding="utf-8")

import sys
sys.path.insert(0, ".")                        # 작업 폴더(data)에 만든 out_flow.py 를 찾게 한다
from out_flow import Stop, run_flow
# 결과: 불러온 함수 이름
print(run_flow.__name__, Stop.__name__)

import pandas as pd

# 작업 함수는 모두 ctx 하나를 받는다. 앞 작업의 결과는 ctx["앞 작업 이름"] 으로 꺼낸다
def read_docs(ctx):
    files = sorted(Path("library/day0").glob("*.txt"))
    return {f.stem: f.read_text(encoding="utf-8") for f in files}


def check(ctx):
    docs = ctx["문서 읽기"]
    empty = [n for n, t in docs.items() if not t.strip()]
    if empty:                                  # 빈 문서가 있으면 고장이 아니라 '멈춤'
        raise Stop(f"빈 문서 {empty}")
    return len(docs)


def split(ctx):
    # 문서를 빈 줄로 나누고, 첫 문단(제목)을 뺀 문단 하나를 조각 하나로 본다
    return [p for t in ctx["문서 읽기"].values() for p in t.split("\n\n")[1:] if p.strip()]


def report(ctx):
    return f"문서 {ctx['검사']}개, 조각 {len(ctx['조각 나누기'])}개"


tasks = {"문서 읽기": (read_docs, []), "검사": (check, ["문서 읽기"]),
         "조각 나누기": (split, ["검사"]), "보고": (report, ["조각 나누기"])}
records, ctx = run_flow(tasks)
# 결과: 작업마다의 상태와 시도 횟수, 보고 작업이 돌려준 글
print(pd.DataFrame(records)[["작업", "상태", "시도"]].to_string(index=False))
print(ctx["보고"])

# 공유 폴더가 첫 번째에만 끊기는 상황을 흉내 낸다(씨앗 대신 횟수로 정해 늘 같게)
calls = {"수": 0}


def flaky_read(ctx):
    calls["수"] += 1
    if calls["수"] == 1:
        raise ConnectionError("공유 폴더 연결이 끊겼습니다")
    return read_docs(ctx)


# 다시 시도 1번(retries=1): 첫 시도가 실패해도 두 번째에 성공한다
records, _ = run_flow(dict(tasks, **{"문서 읽기": (flaky_read, [])}), retries=1)
print(pd.DataFrame(records)[["작업", "상태", "시도"]].to_string(index=False))


# 빈 파일이 섞인 밤(day3)의 문서: 검사가 '멈춤'을 내고 뒤 작업은 건너뛴다
def read_day3(ctx):
    return {f.stem: f.read_text(encoding="utf-8") for f in sorted(Path("library/day3").glob("*.txt"))}


records, _ = run_flow(dict(tasks, **{"문서 읽기": (read_day3, [])}))
# 결과: 검사는 멈춤(다시 시도하지 않는다), 뒤 두 작업은 건너뜀
print(pd.DataFrame(records)[["작업", "상태", "시도", "메모"]].to_string(index=False))