Skip to content

graflo.hq.bulk.ledger

The run ledger: which phases and chunks of a bulk or sync run are done, durably.

One JSON line per event, flushed and synced before the call returns, so after a crash the ledger never claims more than happened. A chunk recorded as started but not committed may or may not be in the target: it is written again, as a replay. A restart forgets every chunk, for a target that lost them.

Attributes

LEDGER = 'ledger.jsonl' module-attribute

Classes

ChunkState

Bases: StrEnum

Source code in graflo/hq/bulk/ledger.py 
33
34
35
36
class ChunkState(StrEnum):
    NEW = "new"
    UNCERTAIN = "uncertain"
    COMMITTED = "committed"

Attributes

COMMITTED = 'committed' class-attribute instance-attribute
NEW = 'new' class-attribute instance-attribute
UNCERTAIN = 'uncertain' class-attribute instance-attribute

Phase

Bases: StrEnum

Phases of a bulk or sync run, in order.

Source code in graflo/hq/bulk/ledger.py 
20
21
22
23
24
25
26
27
28
29
30
class Phase(StrEnum):
    """Phases of a bulk or sync run, in order."""

    OPENED = "opened"
    STAGED = "staged"
    FOLDED = "folded"
    GROUPED = "grouped"
    BEGINNING = "beginning"
    BEGUN = "begun"
    EDGES_PREPARED = "edges_prepared"
    FINISHED = "finished"

Attributes

BEGINNING = 'beginning' class-attribute instance-attribute
BEGUN = 'begun' class-attribute instance-attribute
EDGES_PREPARED = 'edges_prepared' class-attribute instance-attribute
FINISHED = 'finished' class-attribute instance-attribute
FOLDED = 'folded' class-attribute instance-attribute
GROUPED = 'grouped' class-attribute instance-attribute
OPENED = 'opened' class-attribute instance-attribute
STAGED = 'staged' class-attribute instance-attribute

RunLedger

The ledger of the bulk run in staging_dir, read on open.

Source code in graflo/hq/bulk/ledger.py 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
class RunLedger:
    """The ledger of the bulk run in *staging_dir*, read on open."""

    def __init__(self, staging_dir: Path) -> None:
        self.path = Path(staging_dir) / LEDGER
        self._lock = threading.Lock()
        self._phases: set[str] = set()
        self._started: set[str] = set()
        self._committed: set[str] = set()
        self.mode: str | None = None
        if self.path.exists():
            for line in self.path.read_text().splitlines():
                if not line.strip():
                    continue
                event = json.loads(line)
                if "mode" in event:
                    self.mode = event["mode"]
                elif "phase" in event:
                    self._phases.add(event["phase"])
                elif "restart" in event:
                    self._forget()
                elif event.get("state") == "started":
                    self._started.add(event["chunk"])
                elif event.get("state") == "committed":
                    self._committed.add(event["chunk"])

    @property
    def exists(self) -> bool:
        return self.path.exists()

    def has(self, phase: Phase) -> bool:
        return phase.value in self._phases

    def set_mode(self, mode: str) -> None:
        """Record the run's mode, ``bulk`` or ``sync``; first, in a new ledger."""
        self._append({"mode": mode})
        self.mode = mode

    def mark(self, phase: Phase) -> None:
        if phase.value not in self._phases:
            self._append({"phase": phase.value})
            self._phases.add(phase.value)

    def chunk(self, chunk_id: str) -> ChunkState:
        if chunk_id in self._committed:
            return ChunkState.COMMITTED
        if chunk_id in self._started:
            return ChunkState.UNCERTAIN
        return ChunkState.NEW

    def started(self, chunk_id: str) -> None:
        self._append({"chunk": chunk_id, "state": "started"})
        self._started.add(chunk_id)

    def committed(self, chunk_id: str) -> None:
        self._append({"chunk": chunk_id, "state": "committed"})
        self._committed.add(chunk_id)

    def restart(self) -> None:
        """Forget every chunk and the phases after the load began."""
        self._append({"restart": "chunks"})
        self._forget()

    def _forget(self) -> None:
        self._started.clear()
        self._committed.clear()
        self._phases -= {Phase.EDGES_PREPARED.value, Phase.FINISHED.value}

    def _append(self, event: dict[str, str]) -> None:
        line = json.dumps(event, separators=(",", ":")) + "\n"
        with self._lock:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            with open(self.path, "a") as out:
                out.write(line)
                out.flush()
                os.fsync(out.fileno())

Attributes

exists property
mode = None instance-attribute
path = Path(staging_dir) / LEDGER instance-attribute

Methods:

__init__(staging_dir)
Source code in graflo/hq/bulk/ledger.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def __init__(self, staging_dir: Path) -> None:
    self.path = Path(staging_dir) / LEDGER
    self._lock = threading.Lock()
    self._phases: set[str] = set()
    self._started: set[str] = set()
    self._committed: set[str] = set()
    self.mode: str | None = None
    if self.path.exists():
        for line in self.path.read_text().splitlines():
            if not line.strip():
                continue
            event = json.loads(line)
            if "mode" in event:
                self.mode = event["mode"]
            elif "phase" in event:
                self._phases.add(event["phase"])
            elif "restart" in event:
                self._forget()
            elif event.get("state") == "started":
                self._started.add(event["chunk"])
            elif event.get("state") == "committed":
                self._committed.add(event["chunk"])
chunk(chunk_id)
Source code in graflo/hq/bulk/ledger.py 
82
83
84
85
86
87
def chunk(self, chunk_id: str) -> ChunkState:
    if chunk_id in self._committed:
        return ChunkState.COMMITTED
    if chunk_id in self._started:
        return ChunkState.UNCERTAIN
    return ChunkState.NEW
committed(chunk_id)
Source code in graflo/hq/bulk/ledger.py 
93
94
95
def committed(self, chunk_id: str) -> None:
    self._append({"chunk": chunk_id, "state": "committed"})
    self._committed.add(chunk_id)
has(phase)
Source code in graflo/hq/bulk/ledger.py 
69
70
def has(self, phase: Phase) -> bool:
    return phase.value in self._phases
mark(phase)
Source code in graflo/hq/bulk/ledger.py 
77
78
79
80
def mark(self, phase: Phase) -> None:
    if phase.value not in self._phases:
        self._append({"phase": phase.value})
        self._phases.add(phase.value)
restart()

Forget every chunk and the phases after the load began.

Source code in graflo/hq/bulk/ledger.py 
 97
 98
 99
100
def restart(self) -> None:
    """Forget every chunk and the phases after the load began."""
    self._append({"restart": "chunks"})
    self._forget()
set_mode(mode)

Record the run's mode, bulk or sync; first, in a new ledger.

Source code in graflo/hq/bulk/ledger.py 
72
73
74
75
def set_mode(self, mode: str) -> None:
    """Record the run's mode, ``bulk`` or ``sync``; first, in a new ledger."""
    self._append({"mode": mode})
    self.mode = mode
started(chunk_id)
Source code in graflo/hq/bulk/ledger.py 
89
90
91
def started(self, chunk_id: str) -> None:
    self._append({"chunk": chunk_id, "state": "started"})
    self._started.add(chunk_id)