313 lines
12 KiB
Python
313 lines
12 KiB
Python
"""Giving every flow and dashboard stored before ids existed its id, once.
|
|
|
|
Before ids, a flow was its key: ``name`` was its directory, its URL, its hook
|
|
and its rows in the database. The id now does those jobs and the key keeps its
|
|
own, the namespace of the flow's messages (see `fluksio.flow.store`). A store
|
|
from before is moved over by writing each document's key into it as its id.
|
|
Nothing else changes, since every directory, URL, hook, device and row already
|
|
holds that string.
|
|
|
|
`assign_ids` does it when the engine starts, and refuses to touch a store whose
|
|
documents do not say clearly who they are. `plan_ids` is the same reading
|
|
without the writing, which is what ``fluksio store migrate --dry-run`` prints
|
|
and ``fluksio store ls`` lists.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
import subprocess
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from fluksio.flow.dashboards import DASHBOARD_DIR
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
#: The one commit the documents given an id land in.
|
|
COMMIT_MESSAGE = "Assign ids to flows and dashboards"
|
|
|
|
#: Each kind's documents, published first.
|
|
_DOCUMENTS = {
|
|
"flow": ("flow.json", "flow.draft.json"),
|
|
"dashboard": ("dashboard.json", "dashboard.draft.json"),
|
|
}
|
|
|
|
Documents = dict[tuple[str, str], dict[str, dict[str, Any]]]
|
|
|
|
|
|
class IdsRefused(RuntimeError):
|
|
"""A store whose documents do not say clearly enough who they are."""
|
|
|
|
def __init__(self, root: Path, problems: list[str]) -> None:
|
|
super().__init__(root, problems)
|
|
self.root = root
|
|
self.problems = problems
|
|
|
|
def __str__(self) -> str:
|
|
lines = "\n".join(f" {problem}" for problem in self.problems)
|
|
return (
|
|
f"Could not give the flows and dashboards in {self.root} their ids, "
|
|
f"so nothing was changed:\n{lines}\n"
|
|
"Fix these by hand and start again. `fluksio store migrate "
|
|
f"--dry-run {self.root}` checks a store without starting the engine."
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class IdPlan:
|
|
"""What `assign_ids` would do to a store, read without changing it."""
|
|
|
|
#: Every object found, as ``(kind, id, key)``. The id is the directory,
|
|
#: which is also the one a document without an id would be given.
|
|
objects: list[tuple[str, str, str]] = field(default_factory=list)
|
|
#: Documents to give their id, relative to the store, with the version
|
|
#: each moves to.
|
|
stamps: dict[str, int] = field(default_factory=dict)
|
|
#: Documents an interrupted run already gave theirs, not yet committed.
|
|
pending: dict[str, int] = field(default_factory=dict)
|
|
#: Why it refuses, a line each. Nothing is written while there is one.
|
|
problems: list[str] = field(default_factory=list)
|
|
|
|
|
|
def _marker(root: Path) -> Path:
|
|
"""Where a run writes down what it is about to do, before doing any of it.
|
|
|
|
Inside ``.git``, where no commit picks it up, or in the store itself when
|
|
there is no repository, and so no commit either.
|
|
"""
|
|
git = root / ".git"
|
|
return (git if git.is_dir() else root) / ".assign-ids.json"
|
|
|
|
|
|
def _scan(root: Path) -> tuple[Documents, list[str]]:
|
|
"""Every document as plain JSON, by ``(kind, directory)``, and the unreadable."""
|
|
found: Documents = {}
|
|
problems = []
|
|
for kind, base in (("flow", root), ("dashboard", root / DASHBOARD_DIR)):
|
|
for name in _DOCUMENTS[kind]:
|
|
for path in sorted(base.glob(f"*/{name}")):
|
|
directory = path.parent.name
|
|
# `_devices/flow.json` is a device named "flow", and no other
|
|
# `_` directory is a flow either.
|
|
if kind == "flow" and directory.startswith(("_", ".")):
|
|
continue
|
|
rel = path.relative_to(root).as_posix()
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except (OSError, ValueError) as exc:
|
|
problems.append(f"{rel}: not readable as JSON ({exc})")
|
|
continue
|
|
if not isinstance(data, dict) or not isinstance(data.get("name"), str):
|
|
problems.append(f"{rel}: has no name")
|
|
continue
|
|
found.setdefault((kind, directory), {})[rel] = data
|
|
return found, problems
|
|
|
|
|
|
def plan_ids(root: Path) -> IdPlan:
|
|
"""What `assign_ids` would do to the store at *root*, changing nothing.
|
|
|
|
A document without an id is to get its directory as one, which has to be
|
|
its ``name`` as well: the engine runs a flow under the name in its
|
|
document, so when the two differ which one the flow is cannot be told, and
|
|
that is for a person to settle. Refused likewise: a document that is not
|
|
JSON, an id that is not its directory, a published document and draft
|
|
with different keys, and two objects of one kind sharing a key.
|
|
"""
|
|
plan = IdPlan()
|
|
found, plan.problems = _scan(root)
|
|
try:
|
|
recorded = json.loads(_marker(root).read_text())
|
|
except FileNotFoundError:
|
|
recorded = {}
|
|
|
|
owners: dict[tuple[str, str], list[str]] = {}
|
|
for (kind, directory), docs in sorted(found.items()):
|
|
before = len(plan.problems)
|
|
keys = sorted({data["name"] for data in docs.values()})
|
|
for key in keys:
|
|
owners.setdefault((kind, key), []).append(directory)
|
|
if len(keys) > 1:
|
|
plan.problems.append(
|
|
f"{kind} '{directory}': its published document and its draft "
|
|
f"have different keys, '{keys[0]}' and '{keys[1]}'"
|
|
)
|
|
# The working copy's key: the draft's, when there is one.
|
|
plan.objects.append((kind, directory, list(docs.values())[-1]["name"]))
|
|
|
|
unstamped = []
|
|
for rel, data in docs.items():
|
|
given, name = data.get("id"), data["name"]
|
|
if given and given != directory:
|
|
plan.problems.append(f"{rel}: its id '{given}' is not its directory")
|
|
elif not given and name != directory:
|
|
plan.problems.append(
|
|
f"{rel}: its name '{name}' is not its directory '{directory}'"
|
|
)
|
|
elif not isinstance(data.get("version", 1), int):
|
|
plan.problems.append(f"{rel}: its version is not a number")
|
|
elif not given:
|
|
unstamped.append(rel)
|
|
if not unstamped or len(plan.problems) > before:
|
|
continue
|
|
# Past both, so neither an editor on the draft nor one on what a
|
|
# discard goes back to can save over it.
|
|
target = max(data.get("version", 1) for data in docs.values()) + 1
|
|
for rel in unstamped:
|
|
# An interrupted run's number, unless the document moved past it.
|
|
planned = recorded.get(rel, 0)
|
|
own = docs[rel].get("version", 1)
|
|
plan.stamps[rel] = planned if planned > own else target
|
|
|
|
for (kind, key), directories in sorted(owners.items()):
|
|
if len(directories) > 1:
|
|
plan.problems.append(
|
|
f"the {kind}s {' and '.join(map(repr, directories))} share "
|
|
f"the key '{key}'"
|
|
)
|
|
plan.pending = {
|
|
rel: version
|
|
for rel, version in recorded.items()
|
|
if rel not in plan.stamps and (root / rel).exists()
|
|
}
|
|
return plan
|
|
|
|
|
|
def assign_ids(root: Path) -> list[str]:
|
|
"""Give every flow and dashboard at *root* that has no id its key as one.
|
|
|
|
For the engine's startup, with the store to itself and before anything
|
|
reads a definition. Raises `IdsRefused`, having written nothing, on a
|
|
problem `plan_ids` finds. Otherwise each document gets its id and moves
|
|
past the higher version of itself and its sibling, so an editor holding
|
|
the old one gets a conflict, after a discard too. All of it is one commit
|
|
of only those files, whatever else in the store is uncommitted.
|
|
|
|
The versions are written down before the first document is, and the note
|
|
goes once the commit is made, so a run interrupted anywhere in between is
|
|
finished by the next, with the same numbers. A store that has been through
|
|
it changes nothing and says nothing. Returns the documents changed.
|
|
"""
|
|
plan = plan_ids(root)
|
|
if plan.problems:
|
|
raise IdsRefused(root, plan.problems)
|
|
changed = {**plan.pending, **plan.stamps}
|
|
if not changed:
|
|
return []
|
|
|
|
marker = _marker(root)
|
|
if plan.stamps:
|
|
_replace(marker, json.dumps(changed, indent=2) + "\n")
|
|
for rel, version in sorted(plan.stamps.items()):
|
|
path = root / rel
|
|
data = json.loads(path.read_text())
|
|
data.pop("id", None)
|
|
stamped = {"id": path.parent.name, **data, "version": version}
|
|
_replace(path, json.dumps(stamped, indent=2, ensure_ascii=False) + "\n")
|
|
|
|
paths = sorted(changed)
|
|
committed = _commit(root, paths, COMMIT_MESSAGE)
|
|
logger.warning(
|
|
"Gave each flow and dashboard without an id its key as one, in %s",
|
|
", ".join(paths),
|
|
)
|
|
if committed:
|
|
marker.unlink(missing_ok=True)
|
|
return paths
|
|
|
|
|
|
def write_documents(root: Path, texts: dict[str, str], message: str) -> bool:
|
|
"""Write each of *texts*, by path relative to *root*, whose file holds
|
|
something else, then commit those paths alone with *message*.
|
|
|
|
What a key rename writes and undoes (`fluksio.flow.rename`). Whatever else
|
|
is uncommitted stays so. False when the commit failed (`_commit`).
|
|
"""
|
|
for rel, text in texts.items():
|
|
path = root / rel
|
|
if not path.exists() or path.read_text() != text:
|
|
_replace(path, text)
|
|
return _commit(root, sorted(texts), message)
|
|
|
|
|
|
def _replace(path: Path, text: str) -> None:
|
|
"""Write *path* whole or not at all: a crash leaves the old file or the new.
|
|
|
|
The directory is synced too, without which the replace itself may not
|
|
survive a power cut.
|
|
"""
|
|
temp = path.with_name(f".{path.name}.tmp")
|
|
with temp.open("w") as handle:
|
|
handle.write(text)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temp, path)
|
|
directory = os.open(path.parent, os.O_RDONLY)
|
|
try:
|
|
os.fsync(directory)
|
|
finally:
|
|
os.close(directory)
|
|
|
|
|
|
def _commit(root: Path, paths: list[str], message: str) -> bool:
|
|
"""Commit *paths* alone with *message*, leaving whatever else is
|
|
uncommitted as it was.
|
|
|
|
True once nothing is left to commit: done now, done by the run this one
|
|
finishes, or no git to do it with, in which case the store is just files,
|
|
as `FlowStore` takes it too.
|
|
"""
|
|
if not (root / ".git").exists():
|
|
return True
|
|
|
|
def git(*args: str) -> subprocess.CompletedProcess[str]:
|
|
# `gc.auto=0` for the reason `FlowStore._git` gives.
|
|
return subprocess.run(
|
|
["git", "-C", str(root), "-c", "gc.auto=0", *args],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
try:
|
|
git("add", "--", *paths)
|
|
if git("diff", "--cached", "--quiet", "HEAD", "--", *paths).returncode == 0:
|
|
return True
|
|
result = git(
|
|
"-c",
|
|
"user.name=fluksio",
|
|
"-c",
|
|
"user.email=fluksio@localhost",
|
|
"commit",
|
|
"-q",
|
|
"--only",
|
|
"-m",
|
|
message,
|
|
"--",
|
|
*paths,
|
|
)
|
|
except FileNotFoundError:
|
|
return True
|
|
if result.returncode != 0:
|
|
logger.warning(
|
|
"Could not commit '%s': %s",
|
|
message,
|
|
(result.stderr or result.stdout).strip(),
|
|
)
|
|
return False
|
|
return True
|
|
|
|
|
|
__all__ = [
|
|
"COMMIT_MESSAGE",
|
|
"IdPlan",
|
|
"IdsRefused",
|
|
"assign_ids",
|
|
"plan_ids",
|
|
"write_documents",
|
|
]
|