Class tigrbl.op.model_registry.OpspecRegistry

tigrbl.op.model_registry.OpspecRegistry

OpspecRegistry(table)

Per-model OpSpec registry with change notifications.

• Stores specs keyed by (alias, target).
• Adds/sets/removes specs and notifies listeners with the changed keys.
• Binder should call subscribe(...) to rebuild a model's namespaces when the registry changes (partial rebuild is possible based on changed keys).

Thread-safe via an instance-level RLock.

Source code in tigrbl/op/model_registry.py

def __init__(self, table: type) -> None:
    self._table: type = table
    self._items: Dict[Tuple[str, TargetOp], OpSpec] = {}
    self._lock = RLock()
    # store weakrefs to listener callables where possible; fallback to strong refs
    self._listeners: List[
        Callable[[OpspecRegistry, set[Tuple[str, TargetOp]]], None]
    ] = []
    self._version: int = 0

table property

table

version property

version

keys

keys()

Source code in tigrbl/op/model_registry.py

def keys(self) -> Iterator[Tuple[str, TargetOp]]:
    with self._lock:
        return iter(tuple(self._items.keys()))

items

items()

Source code in tigrbl/op/model_registry.py

def items(self) -> Iterator[Tuple[Tuple[str, TargetOp], OpSpec]]:
    with self._lock:
        return iter(tuple(self._items.items()))

values

values()

Source code in tigrbl/op/model_registry.py

def values(self) -> Iterator[OpSpec]:
    with self._lock:
        return iter(tuple(self._items.values()))

get_all

get_all()

Stable snapshot of all specs.

Source code in tigrbl/op/model_registry.py

def get_all(self) -> Tuple[OpSpec, ...]:
    """Stable snapshot of all specs."""
    with self._lock:
        return tuple(self._items.values())

subscribe

subscribe(fn)

Register a listener to be called on changes. NOTE: The listener should be idempotent. It receives (registry, changed_keys).

Source code in tigrbl/op/model_registry.py

def subscribe(self, fn: Listener) -> None:
    """
    Register a listener to be called on changes.
    NOTE: The listener should be idempotent. It receives (registry, changed_keys).
    """
    with self._lock:
        # Avoid duplicate subscriptions
        if fn not in self._listeners:
            self._listeners.append(fn)

unsubscribe

unsubscribe(fn)

Source code in tigrbl/op/model_registry.py

def unsubscribe(self, fn: Listener) -> None:
    with self._lock:
        try:
            self._listeners.remove(fn)
        except ValueError:
            pass

add

add(specs)

Add or overwrite one or more specs. Returns the set of changed keys.

Source code in tigrbl/op/model_registry.py

def add(self, specs: Iterable[OpSpec] | OpSpec) -> set[Tuple[str, TargetOp]]:
    """
    Add or overwrite one or more specs.
    Returns the set of changed keys.
    """
    if isinstance(specs, OpSpec):
        specs = (specs,)

    changed: set[Tuple[str, TargetOp]] = set()
    with self._lock:
        for sp in specs:
            sp = _ensure_table(sp, self._table)
            k = _spec_key(sp)
            if self._items.get(k) is sp:
                continue  # exact object already present
            self._items[k] = sp
            changed.add(k)
        if changed:
            self._version += 1
    if changed:
        self._notify(changed)
    return changed

set

set(specs)

Replace all specs with the provided iterable. Returns the set of changed keys (union of removed + added/updated).

Source code in tigrbl/op/model_registry.py

def set(self, specs: Iterable[OpSpec]) -> set[Tuple[str, TargetOp]]:
    """
    Replace all specs with the provided iterable.
    Returns the set of changed keys (union of removed + added/updated).
    """
    new_map: Dict[Tuple[str, TargetOp], OpSpec] = {}
    for sp in specs:
        sp = _ensure_table(sp, self._table)
        new_map[_spec_key(sp)] = sp

    with self._lock:
        old_keys = set(self._items.keys())
        new_keys = set(new_map.keys())

        removed = old_keys - new_keys
        added_or_updated = {
            k for k in new_keys if self._items.get(k) is not new_map[k]
        }

        changed = removed | added_or_updated
        self._items = new_map
        if changed:
            self._version += 1

    if changed:
        self._notify(changed)
    return changed

remove

remove(alias, target=None)

Remove specs by alias (optionally constrain to a specific target). Returns the set of removed keys.

Source code in tigrbl/op/model_registry.py

def remove(
    self, alias: str, target: TargetOp | None = None
) -> set[Tuple[str, TargetOp]]:
    """
    Remove specs by alias (optionally constrain to a specific target).
    Returns the set of removed keys.
    """
    removed: set[Tuple[str, TargetOp]] = set()
    with self._lock:
        if target is None:
            # remove all targets under this alias
            for k in list(self._items.keys()):
                if k[0] == alias:
                    self._items.pop(k, None)
                    removed.add(k)
        else:
            k = (alias, target)
            if k in self._items:
                self._items.pop(k, None)
                removed.add(k)

        if removed:
            self._version += 1

    if removed:
        self._notify(removed)
    return removed

clear

clear()

Source code in tigrbl/op/model_registry.py

def clear(self) -> None:
    with self._lock:
        if not self._items:
            return
        self._items.clear()
        self._version += 1
    self._notify(set())