2.27.2.6. pycsamt.pipeline._history

Append-only run history log for Pipeline.run(..., history=...).

One compact JSON line per run — not full StepResult objects — so the log stays greppable and cheap to append/read even across many runs. This is deliberately a plain JSON-lines file, not a database: it exists so a user can answer “how did my last N runs compare,” not to be a query engine.

Functions

append_run(path, result)	Append one JSON line summarizing result to path.
default_history_path()
load_history([path, last])	Read back previously logged run summaries, oldest first.

pycsamt.pipeline._history.append_run(path, result)

Append one JSON line summarizing result to path.

Creates parent directories as needed. Never raises on a write failure when called from Pipeline.run() — that call site wraps this in its own try/except and warns instead, the same “a logging feature must not abort the run” posture as every other optional side effect in the run loop (QC plots, intermediate snapshots).

Parameters:

• path (str | Path)

• result (PipelineResult)

Return type:

None

pycsamt.pipeline._history.load_history(path=None, *, last=None)

Read back previously logged run summaries, oldest first.

Parameters:

• path (str | Path | None) – Defaults to default_history_path().

• last (int | None) – When given, return only the most recent last entries.

• skipped (A malformed line (e.g. from a write interrupted mid-flush) is)

• non-event (rather than raising — the same "one bad entry is a)

• a (not)

Return type:

list[dict]

:param crash” posture StepCache already uses.:

pycsamt.pipeline._history.default_history_path()

Return type:

Path