A namespaced key/value view for scheduling state that outlives a request.
Multi-user scheduling needs a little state to persist between the independent
calls that serve each user (who holds which slot). [Scope][] is the small
interface that bookkeeping like
Reservations depends
on, so the scheduling library never imports a server: the server backs a
Scope with SQLite, while tests use the in-memory [DictScope][].
A scope is a string-keyed store of JSON-native values plus a scope(name)
child namespace, so independent consumers never collide on key names.
Scope state is regenerable bookkeeping, never an archive: every consumer can
rebuild what it needs from a fresh scope on the next request. That is what lets
[ensure_format][] discard state written in an outdated shape instead of
migrating it.
DictScope
An in-memory [Scope][] backed by plain dicts.
The reference implementation and the test double: a server provides a
durable, cross-process scope (SQLite), while a unit test constructs a
DictScope and reads the resulting state straight back out.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| class DictScope:
"""An in-memory [Scope][] backed by plain dicts.
The reference implementation and the test double: a server provides a
durable, cross-process scope (SQLite), while a unit test constructs a
``DictScope`` and reads the resulting state straight back out.
"""
def __init__(
self,
values: dict[str, JSONValue] | None = None,
children: dict[str, "DictScope"] | None = None,
) -> None:
"""Wrap ``values`` and ``children``, or start empty if omitted."""
self._values: dict[str, JSONValue] = {} if values is None else values
self._children: dict[str, DictScope] = {} if children is None else children
def __getitem__(self, key: str) -> JSONValue:
"""Return the value stored under ``key``, raising if absent."""
return self._values[key]
def __setitem__(self, key: str, value: JSONValue) -> None:
"""Store ``value`` under ``key``."""
self._values[key] = value
def __delitem__(self, key: str) -> None:
"""Remove ``key`` and its value from this scope."""
del self._values[key]
def __contains__(self, key: str) -> bool:
"""Return whether ``key`` is present in this scope."""
return key in self._values
def get(self, key: str, default: JSONValue = None) -> JSONValue:
"""Return the value stored under ``key``, or ``default`` if absent."""
return self._values.get(key, default)
def items(self) -> list[tuple[str, JSONValue]]:
"""Return a materialized snapshot of this scope's key/value pairs."""
return list(self._values.items())
def scope(self, name: str) -> "DictScope":
"""Return the child namespace ``name``, creating it if needed."""
if name not in self._children:
self._children[name] = DictScope()
return self._children[name]
def clear(self) -> None:
"""Delete every value in this scope and every descendant namespace."""
self._values.clear()
self._children.clear()
|
__contains__(key)
Return whether key is present in this scope.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __contains__(self, key: str) -> bool:
"""Return whether ``key`` is present in this scope."""
return key in self._values
|
__delitem__(key)
Remove key and its value from this scope.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __delitem__(self, key: str) -> None:
"""Remove ``key`` and its value from this scope."""
del self._values[key]
|
__getitem__(key)
Return the value stored under key, raising if absent.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __getitem__(self, key: str) -> JSONValue:
"""Return the value stored under ``key``, raising if absent."""
return self._values[key]
|
__init__(values=None, children=None)
Wrap values and children, or start empty if omitted.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __init__(
self,
values: dict[str, JSONValue] | None = None,
children: dict[str, "DictScope"] | None = None,
) -> None:
"""Wrap ``values`` and ``children``, or start empty if omitted."""
self._values: dict[str, JSONValue] = {} if values is None else values
self._children: dict[str, DictScope] = {} if children is None else children
|
__setitem__(key, value)
Store value under key.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __setitem__(self, key: str, value: JSONValue) -> None:
"""Store ``value`` under ``key``."""
self._values[key] = value
|
clear()
Delete every value in this scope and every descendant namespace.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def clear(self) -> None:
"""Delete every value in this scope and every descendant namespace."""
self._values.clear()
self._children.clear()
|
get(key, default=None)
Return the value stored under key, or default if absent.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def get(self, key: str, default: JSONValue = None) -> JSONValue:
"""Return the value stored under ``key``, or ``default`` if absent."""
return self._values.get(key, default)
|
items()
Return a materialized snapshot of this scope's key/value pairs.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def items(self) -> list[tuple[str, JSONValue]]:
"""Return a materialized snapshot of this scope's key/value pairs."""
return list(self._values.items())
|
scope(name)
Return the child namespace name, creating it if needed.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def scope(self, name: str) -> "DictScope":
"""Return the child namespace ``name``, creating it if needed."""
if name not in self._children:
self._children[name] = DictScope()
return self._children[name]
|
Scope
Bases: Protocol
A namespaced key/value view of persisted, JSON-native state.
Values must be JSON-native (the concrete backing store serializes them);
consumers that hold richer values encode them first (e.g. an instant as an
ISO-8601 string). scope(name) returns an isolated child namespace.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| @runtime_checkable
class Scope(Protocol):
"""A namespaced key/value view of persisted, JSON-native state.
Values must be JSON-native (the concrete backing store serializes them);
consumers that hold richer values encode them first (e.g. an instant as an
ISO-8601 string). ``scope(name)`` returns an isolated child namespace.
"""
def __getitem__(self, key: str) -> JSONValue:
"""Return the value stored under ``key``, raising if absent."""
...
def __setitem__(self, key: str, value: JSONValue) -> None:
"""Store ``value`` under ``key``."""
...
def __delitem__(self, key: str) -> None:
"""Remove ``key`` and its value from this scope."""
...
def __contains__(self, key: str) -> bool:
"""Return whether ``key`` is present in this scope."""
...
def get(self, key: str, default: JSONValue = None) -> JSONValue:
"""Return the value stored under ``key``, or ``default`` if absent."""
...
def items(self) -> list[tuple[str, JSONValue]]:
"""A materialized snapshot, so callers may delete keys while iterating."""
...
def scope(self, name: str) -> "Scope":
"""Return the child namespace ``name``, creating it if needed."""
...
def clear(self) -> None:
"""Delete every value in this scope and every descendant namespace."""
...
|
__contains__(key)
Return whether key is present in this scope.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __contains__(self, key: str) -> bool:
"""Return whether ``key`` is present in this scope."""
...
|
__delitem__(key)
Remove key and its value from this scope.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __delitem__(self, key: str) -> None:
"""Remove ``key`` and its value from this scope."""
...
|
__getitem__(key)
Return the value stored under key, raising if absent.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __getitem__(self, key: str) -> JSONValue:
"""Return the value stored under ``key``, raising if absent."""
...
|
__setitem__(key, value)
Store value under key.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def __setitem__(self, key: str, value: JSONValue) -> None:
"""Store ``value`` under ``key``."""
...
|
clear()
Delete every value in this scope and every descendant namespace.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def clear(self) -> None:
"""Delete every value in this scope and every descendant namespace."""
...
|
get(key, default=None)
Return the value stored under key, or default if absent.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def get(self, key: str, default: JSONValue = None) -> JSONValue:
"""Return the value stored under ``key``, or ``default`` if absent."""
...
|
items()
A materialized snapshot, so callers may delete keys while iterating.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def items(self) -> list[tuple[str, JSONValue]]:
"""A materialized snapshot, so callers may delete keys while iterating."""
...
|
scope(name)
Return the child namespace name, creating it if needed.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def scope(self, name: str) -> "Scope":
"""Return the child namespace ``name``, creating it if needed."""
...
|
Reset scope unless its state was written in format version.
A consumer that persists structured values stamps its scope with the format
version it writes. When the shape of what it persists changes, it bumps the
version, and any scope stamped otherwise — including unstamped state from
before the consumer existed in its current form — is cleared and rebuilt by
the requests that follow. State is discarded rather than migrated: scopes
hold regenerable bookkeeping, so a reset costs at most one recomputation.
Source code in capturegraph-lib/capturegraph/scheduling/organize/scope.py
| def ensure_format(scope: Scope, version: int) -> None:
"""Reset ``scope`` unless its state was written in format ``version``.
A consumer that persists structured values stamps its scope with the format
version it writes. When the shape of what it persists changes, it bumps the
version, and any scope stamped otherwise — including unstamped state from
before the consumer existed in its current form — is cleared and rebuilt by
the requests that follow. State is discarded rather than migrated: scopes
hold regenerable bookkeeping, so a reset costs at most one recomputation.
"""
if scope.get("format") != version:
scope.clear()
scope["format"] = version
|