class documentation

class ChangeTracker:

Constructor: ChangeTracker(schema_dict)

View In Hierarchy

Owns a schema's op-log, baseline snapshot, and atomic-rollback transaction.

A tracker is bound to a single schema dict at construction and snapshots it as the baseline. Thereafter the owning ~datagraphs.schema.Schema drives it: it wraps each public mutation in track + atomic, journals touched state with the stage* helpers, and calls record exactly once per successful outermost mutation. The accumulated baseline and change_log are then handed to schema_report.build_change_report to produce the changelog.

Method __init__ Bind the tracker to schema_dict and capture it as the baseline.
Method atomic All-or-nothing guard for a multi-step mutation.
Method record Append a single op-log entry to the change log.
Method stage Journal a whole class dict's content for atomic rollback.
Method stage_prop Journal a single property dict's content for atomic rollback.
Method track Re-entrancy depth guard for the op-log.
Method track_added_prop Journal a freshly-appended property so rollback can remove it (O(1)).
Property baseline The schema snapshot captured at construction — the diff's before.
Property change_log The recorded op-log: one entry per successful outermost mutation.
Property depth Current re-entrancy depth (0 when no mutation is in progress).
Static Method _capture_baseline Return an independent deep copy of schema_dict.
Instance Variable _baseline Undocumented
Instance Variable _change_log Undocumented
Instance Variable _schema Undocumented
Instance Variable _staged_classes Undocumented
Instance Variable _staged_props Undocumented
Instance Variable _tracking_depth Undocumented
Instance Variable _undo Undocumented
def __init__(self, schema_dict: dict):

Bind the tracker to schema_dict and capture it as the baseline.

Parameters
schema_dict:dictThe live schema dict to track and guard. The tracker keeps this reference so atomic can roll back its classes list in place.
@contextmanager
def atomic(self, outermost: bool) -> Generator[None, None, None]:

All-or-nothing guard for a multi-step mutation.

At the OUTERMOST public-call boundary (outermost=True) this opens a rollback transaction over the mutable model (schema_dict["classes"] — the only state any public mutating method touches) and, on ANY exception from the body, restores it before re-raising, so no mid-apply raise can leave a partial write. This guard wraps EVERY public mutating method (not just the property create/update paths), making the whole mutating surface all-or-nothing. Combined with the success-only record (ADR 0002), a rolled-back op records nothing, so change_report never surfaces a change for an operation the caller saw raise.

Footprint-scoped, undo-journalled rollback. Rollback state is captured so its cost is proportional to what the operation actually touches, not to the size of the whole schema:

  • a SHALLOW copy of the class list (list(classes)) captures membership, order and identity, so any append / removal / reordering of classes is undoable in O(C) cheap references — no per-class deep copy; and
  • per-touch CONTENT is journalled lazily at the FINEST granularity the mutation needs — a single appended property (track_added_prop, O(1) undo), a single in-place-modified property (stage_prop, deep copy of just that small prop dict), or a whole class dict's scalar/structure (stage, deep copy of one class) for class-level edits.

Granularity matters on the cascade hot path: create_property / update_property with apply_to_subclasses=True touch one property per class across the whole hierarchy. Journalling that single property per target is O(1), so a cascade is O(C) regardless of how many properties each class already carries. The prior design deep-copied each entire class dict (O(properties)), making a sequence of L wide cascades on C classes cost O(L²·C); property-granular journalling makes it O(L·C). Likewise, building an N-class schema one mutation at a time is O(N), not the O(N²) of snapshotting the whole class list per mutation. The O(descendants) cascade asymptotics are unchanged.

On rollback the journal is replayed in REVERSE (so later edits undo before earlier ones), then list membership/order/identity is restored by slice-assignment (not rebinding), so any externally-held reference to the classes list (e.g. via the public classes view or a prior to_dict()) stays consistent with the rolled-back model. In-place restores (clear + update) preserve dict identity, so references to individual class/property dicts also stay valid.

When outermost is False this is an inert pass-through: the outermost frame already owns the transaction for the whole re-entrant chain, so nested/cascade internals neither snapshot nor journal twice.

def record(self, op: str, **args):

Append a single op-log entry to the change log.

Parameters
op:strThe operation name (e.g. "create_class").
**argsKeyword arguments carrying the intent-bearing parameters for the operation (names, flags — not field values).
def stage(self, class_def: dict):

Journal a whole class dict's content for atomic rollback.

Call this BEFORE the first class-level mutation of class_def (a change to a scalar key such as name / subClassOf / labelProperty / description, or a reordering/removal within its properties list). The first call for a given class dict deep-copies it; later calls for the same dict are no-ops, so the cost is one deep copy per distinct class the operation edits at class level.

For the property-cascade hot path prefer the finer-grained track_added_prop / stage_prop, which journal a single property rather than the whole (potentially property-heavy) class dict.

Outside an atomic transaction (_undo is None — e.g. a nested call whose outermost frame has already closed) this is an inert no-op. Newly-appended class dicts need not be staged: the shallow structural snapshot in atomic drops them on rollback regardless.

def stage_prop(self, prop_def: dict):

Journal a single property dict's content for atomic rollback.

Call this BEFORE the first in-place mutation of prop_def (the cascade-update hot path). Deep-copies only the small property dict, not its owning class, so an N-target cascade costs O(N) rather than O(N · properties-per-class). Deduped per property dict; inert outside an atomic transaction.

@contextmanager
def track(self) -> Generator[bool, None, None]:

Re-entrancy depth guard for the op-log.

Yields outermost=True only when entered at depth 0 (i.e. this is the outermost public call in a re-entrant chain). The depth counter is always restored in finally so exceptions cannot leave it incremented.

Usage:

with tracker.track() as outermost, tracker.atomic(outermost):
    # ... do the real work ...
    if outermost:
        tracker.record("op_name", arg1=val1)
def track_added_prop(self, props_list: list, prop_def: dict):

Journal a freshly-appended property so rollback can remove it (O(1)).

Call this immediately after appending prop_def to props_list (the cascade-create hot path). Recording the append rather than deep-copying the owning class keeps a wide create-cascade O(C); on rollback the prop is simply removed. Inert outside an atomic transaction.

@property
baseline: dict =

The schema snapshot captured at construction — the diff's before.

@property
change_log: list[dict] =

The recorded op-log: one entry per successful outermost mutation.

@property
depth: int =

Current re-entrancy depth (0 when no mutation is in progress).

@staticmethod
def _capture_baseline(schema_dict: dict) -> dict:

Return an independent deep copy of schema_dict.

Uses the json round-trip idiom, which relies on the schema dict being JSON-serialisable — an invariant the schema class already establishes.

Parameters
schema_dict:dictThe schema dict to snapshot.
Returns
dictAn independent deep copy.
_baseline: dict =

Undocumented

_change_log: list[dict] =

Undocumented

_schema =

Undocumented

_staged_classes: set[int] | None =

Undocumented

_staged_props: set[int] | None =

Undocumented

_tracking_depth: int =

Undocumented

_undo: list[tuple] | None =

Undocumented