flayer.core.state

Module source

flayer.core.state

Minimal owned-resource snapshots with strict reads and atomic local persistence.

MAX_JSON_BYTES module-attribute

MAX_JSON_BYTES = 1024 * 1024

StateError

Bases: ContractError

Owned state is malformed, mismatched, busy, or cannot be persisted safely.

ResourceState dataclass

ResourceState(logical_id: str, kind: str, resource_id: str)

Minimum provider locator for a resource recorded by one managed stack.

logical_id instance-attribute

logical_id: str

kind instance-attribute

kind: str

resource_id instance-attribute

resource_id: str

StackState dataclass

StackState(identity: StackIdentity, resources: tuple[ResourceState, ...] = (), schema_version: int = SCHEMA_VERSION)

Immutable resource inventory scoped to a complete ownership identity.

identity instance-attribute

identity: StackIdentity

resources class-attribute instance-attribute

resources: tuple[ResourceState, ...] = ()

schema_version class-attribute instance-attribute

schema_version: int = SCHEMA_VERSION

ParseState

ParseState(data: Mapping[str, object], expected_identity: StackIdentity) -> StackState

Decode a versioned snapshot and require explicit caller ownership expectations.

源代码位于： installed/flayer/core/state.py

def ParseState(data: Mapping[str, object], expected_identity: StackIdentity) -> StackState:
    """Decode a versioned snapshot and require explicit caller ownership expectations."""

    try:
        ValidateFields(
            data, frozenset({"schema_version", "identity", "resources"}), frozenset(), "state"
        )
        ValidateSchemaVersion(data["schema_version"])
        identity = ParseIdentity(data["identity"])
        _RequireIdentity(identity, expected_identity)
        raw_resources = data["resources"]

        if not isinstance(raw_resources, list):
            raise ContractError("resources must be an array of tables")

        resources: list[ResourceState] = []

        for item in raw_resources:
            table = RequireTable(item, "resources")
            ValidateFields(
                table,
                frozenset({"logical_id", "kind", "resource_id"}),
                frozenset(),
                "resources",
            )
            resources.append(
                ResourceState(
                    logical_id=ValidateName(table["logical_id"], "resources.logical_id"),
                    kind=ValidateName(table["kind"], "resources.kind"),
                    resource_id=ValidateIdentifier(table["resource_id"], "resources.resource_id"),
                )
            )

        return StackState(identity=identity, resources=tuple(resources))

    except ContractError as error:
        raise StateError(str(error)) from error

LoadState

LoadState(path: str | Path, expected_identity: StackIdentity) -> StackState | None

Return a strictly owned snapshot, or None only when the file is absent.

源代码位于： installed/flayer/core/state.py

def LoadState(path: str | Path, expected_identity: StackIdentity) -> StackState | None:
    """Return a strictly owned snapshot, or None only when the file is absent."""

    state_path = Path(path)

    if not isinstance(expected_identity, StackIdentity):
        raise StateError("expected_identity must be a StackIdentity")

    try:
        _ValidatePath(state_path)
        data = _ReadOwnedJson(state_path)

    except FileNotFoundError:
        return None

    except StateError:
        raise

    except (OSError, ValueError, RecursionError) as error:
        raise StateError("Unable to read a valid state snapshot") from error

    try:
        return ParseState(RequireTable(data, "state"), expected_identity)

    except ContractError as error:
        raise StateError(str(error)) from error

SaveState

SaveState(path: str | Path, state: StackState, expected_identity: StackIdentity) -> None

Atomically save owned state; never replace a foreign or malformed existing snapshot.

源代码位于： installed/flayer/core/state.py

def SaveState(path: str | Path, state: StackState, expected_identity: StackIdentity) -> None:
    """Atomically save owned state; never replace a foreign or malformed existing snapshot."""

    if not isinstance(state, StackState):
        raise StateError("state must be a StackState")

    _RequireIdentity(state.identity, expected_identity)
    serialized = json.dumps(asdict(state), indent=2, sort_keys=True) + "\n"

    if len(serialized.encode("utf-8")) > MAX_JSON_BYTES:
        raise StateError("Owned state snapshot exceeds its bounded JSON storage limit")

    state_path = Path(path)
    temporary_path: Path | None = None

    with _StateLock(state_path):
        LoadState(state_path, expected_identity)

        try:
            descriptor, temporary_name = tempfile.mkstemp(
                prefix=f".{state_path.name}.", suffix=".tmp", dir=state_path.parent
            )
            temporary_path = Path(temporary_name)

            with _OpenTextStream(descriptor, "w") as stream:
                stream.write(serialized)
                stream.flush()
                os.fsync(stream.fileno())

            os.replace(temporary_path, state_path)
            temporary_path = None
            _SyncDirectory(state_path.parent)

        except OSError as error:
            raise StateError(
                "Unable to persist state; a completed replacement may be visible"
            ) from error

        finally:
            if temporary_path is not None:
                try:
                    temporary_path.unlink(missing_ok=True)

                except OSError as error:
                    raise StateError(
                        "Unable to clean temporary state; "
                        "incomplete local cleanup requires recovery"
                    ) from error

RemoveState

RemoveState(path: str | Path, expected_identity: StackIdentity) -> bool

Remove only a matching local snapshot; never delete infrastructure resources.

源代码位于： installed/flayer/core/state.py

def RemoveState(path: str | Path, expected_identity: StackIdentity) -> bool:
    """Remove only a matching local snapshot; never delete infrastructure resources."""

    state_path = Path(path)

    if not isinstance(expected_identity, StackIdentity):
        raise StateError("expected_identity must be a StackIdentity")

    try:
        _ValidatePath(state_path)

        if not state_path.parent.exists():
            return False

        with _StateLock(state_path):
            if LoadState(state_path, expected_identity) is None:
                return False

            state_path.unlink()
            _SyncDirectory(state_path.parent)

    except OSError as error:
        raise StateError("Unable to remove the owned state snapshot") from error

    return True