@click.command("merge")
@click.argument(
"left",
required=False,
type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.argument(
"right",
required=False,
type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
"--op",
"op_path",
type=click.Path(exists=True, dir_okay=False, path_type=Path),
default=None,
help=(
"MergeManifestsOp document: vertex/property/relation equivalences "
"(with the merged key as `identity`) and canonical maps. Omitted "
"merges a disjoint union."
),
)
@click.option(
"-o",
"--output",
type=click.Path(dir_okay=False, path_type=Path),
default=None,
help="Where to write the merged manifest. Omitted prints a summary only.",
)
@click.option(
"--canonical-map",
"canonical_map_options",
multiple=True,
metavar="SIDE=PATH",
help=(
"Canonical map for one side (or `both`), repeatable. Names the "
"merged classes and is checked for disagreement with the op; an "
"equivalence may then name a class by its own or its canonical name."
),
)
@click.option(
"--name-conflict",
type=click.Choice(["error", "prefix_right", "union_right"]),
default=None,
help=(
"Override the op's name_conflict policy: error refuses a name both "
"sides carry and prints the equivalences to declare; union_right "
"unions classes of exactly the same name (each pair becomes a 1-1 "
"equivalence); prefix_right keeps them apart under r_ names."
),
)
@click.option(
"--bump-version",
type=click.Choice(["minor", "none"]),
default="minor",
show_default=True,
help="Bump the merged schema version. 'none' leaves the left's.",
)
@click.option(
"--strict-references",
is_flag=True,
help="Fail on ingestion/bindings references the merged schema lacks.",
)
@click.option(
"--dry-run",
is_flag=True,
help="Merge and report, but write nothing.",
)
@click.option(
"--plot",
"plot_path",
type=click.Path(dir_okay=False, path_type=Path),
default=None,
help=(
"Draw the declaration graph and its conflicts here; the suffix picks "
"the format (svg, pdf, png, dot). Written even when merge refuses."
),
)
@click.option(
"--preview-json",
"preview_json_path",
type=click.Path(dir_okay=False, path_type=Path),
default=None,
help=(
"Write the same preview as JSON: nodes, edges, clusters, findings and "
"the outcome. Written even when merge refuses."
),
)
@click.option(
"--plot-dir",
type=click.Path(file_okay=False, path_type=Path),
default=None,
help=(
"Draw each equivalence here as four small figures (members, attributes, "
"identity, lookups), plus index.json and summary.md. Written even when "
"merge refuses."
),
)
@click.option(
"--plot-format",
type=click.Choice(["svg", "png", "pdf", "dot"]),
default=None,
help="Format of the --plot-dir figures. [default: svg]",
)
@click.option(
"--plot-dpi",
type=click.IntRange(min=1),
default=None,
help="Resolution of --plot-dir figures in png.",
)
@click.option(
"--unit",
"unit_patterns",
multiple=True,
metavar="NAME",
help=(
"Draw only these equivalences, repeatable: a merged name, a position in "
"vertex_equivalences, or a pattern on the cluster id."
),
)
@click.option(
"--view",
"view_names",
multiple=True,
type=click.Choice(["members", "attributes", "identity", "lookups"]),
help="Draw only these views, repeatable. [default: all four]",
)
@click.option(
"--declared",
is_flag=True,
help=(
"Draw the --op declarations alone, without LEFT and RIGHT: no checks, "
"no merge. Takes --plot-dir and --preview-json."
),
)
@click.option(
"--suggest",
"suggest_path",
type=click.Path(dir_okay=False, path_type=Path),
default=None,
help=(
"Write a merge op that settles every problem a rename or an `into` "
"can settle, without guessing, and exit; without --op, a scaffold "
"declaring every name both sides share. Review it before use."
),
)
@click.option(
"--max-rows",
type=click.IntRange(min=0),
default=12,
show_default=True,
help="Attribute rows to draw per class before the rest are summarised.",
)
@click.option(
"--check-profile",
"profile_name",
default=None,
help=(
"Also check the merged manifest against this conformance profile "
"and print the report. Findings do not change the exit code."
),
)
@store_option
@click.option(
"-m",
"--label",
"record_label",
default=None,
help=(
"Record the merge in the store as a two-parent commit under this "
"label. Both inputs must already be in the history; without this the "
"verb writes only the manifest, as before."
),
)
def merge(
left: Path | None,
right: Path | None,
op_path: Path | None,
output: Path | None,
canonical_map_options: tuple[str, ...],
name_conflict: str | None,
bump_version: str,
strict_references: bool,
dry_run: bool,
plot_path: Path | None,
preview_json_path: Path | None,
plot_dir: Path | None,
plot_format: str | None,
plot_dpi: int | None,
unit_patterns: tuple[str, ...],
view_names: tuple[str, ...],
declared: bool,
suggest_path: Path | None,
max_rows: int,
profile_name: str | None,
store: Path,
record_label: str | None,
) -> None:
"""Merge LEFT and RIGHT into one manifest."""
canonical_map_paths = _parse_canonical_map_option(canonical_map_options)
if plot_dir is None:
given = [
name
for name, value in (
("--plot-format", plot_format),
("--plot-dpi", plot_dpi),
("--unit", unit_patterns),
("--view", view_names),
)
if value
]
if given:
raise click.UsageError(f"{', '.join(given)} needs --plot-dir")
units = _Units(
directory=plot_dir,
output_format=plot_format or "svg",
dpi=plot_dpi,
select=unit_patterns,
views=view_names,
)
if declared:
_declared(
left=left,
right=right,
op_path=op_path,
canonical_map_paths=canonical_map_paths,
name_conflict=name_conflict,
units=units,
json_path=preview_json_path,
refused={
"-o": output,
"--plot": plot_path,
"--suggest": suggest_path,
"--dry-run": dry_run,
"--check-profile": profile_name,
"-m": record_label,
},
)
return
if left is None or right is None:
raise click.UsageError("missing LEFT and RIGHT (or pass --declared)")
try:
left_manifest = load_manifest(left)
right_manifest = load_manifest(right)
except (ValueError, TypeError) as exc:
raise _MergeSetupError(
f"not a valid manifest -- {type(exc).__name__}: {exc}"
) from exc
op = _load_op(op_path, name_conflict, canonical_map_paths)
if suggest_path is not None:
_write_suggestion(
suggest_path,
left_manifest,
right_manifest,
op if op_path is not None else None,
)
return
wants_preview = (
plot_path is not None
or preview_json_path is not None
or plot_dir is not None
or dry_run
)
if plot_path is not None:
_check_plot_suffix(plot_path)
# Built before merging and without merging again: `attempt=False`
# keeps this to one merge per invocation, and the outcome is folded in
# below whichever way that one goes.
preview = (
preview_merge(left_manifest, right_manifest, op, attempt=False)
if wants_preview
else None
)
if preview is not None:
units.check(preview)
def emit(outcome: MergeOutcome, subjects: tuple[str, ...] = ()) -> None:
if preview is not None:
write(preview.with_outcome(outcome, subjects=subjects))
def write(attempted: MergePreview) -> None:
_write_preview(
attempted,
plot_path=plot_path,
json_path=preview_json_path,
max_rows=max_rows,
units=units,
)
try:
merged, report = merge_manifests_with_report(
left_manifest,
right_manifest,
op,
bump_version="minor" if bump_version == "minor" else False,
strict_references=strict_references,
)
except MergeIncompleteError as exc:
# Consistent but not covering every name: the completion is the
# declaration to paste into the op, so print it as one.
emit(*outcome_from_exception(exc))
click.echo(f"merge refused: {type(exc).__name__}: {exc}", err=True)
if exc.completion is not None:
click.echo("completion:", err=True)
click.echo(
yaml.safe_dump(exc.completion.to_dict(), sort_keys=False).rstrip(),
err=True,
)
_echo_naming(left_manifest, right_manifest, op, err=True)
raise SystemExit(EXIT_REFUSED)
except (
AlignmentConflictError,
MergeCanonicalConflictError,
MergeIdentityError,
MergeNameConflictError,
ValueError,
) as exc:
# Every one of these carries what to declare next; a traceback would
# bury it.
emit(*outcome_from_exception(exc))
click.echo(f"merge refused: {type(exc).__name__}: {exc}", err=True)
if isinstance(exc, MergeCanonicalConflictError):
_echo_naming(left_manifest, right_manifest, op, err=True)
raise SystemExit(EXIT_REFUSED)
routing = routing_rows(merged, report)
if preview is not None:
write(with_merge_result(preview, merged, report))
for line in _summary(merged):
click.echo(line)
if profile_name is not None:
# The model, not a re-serialization of it. A merged manifest has no
# authored document -- and neither serialization is a substitute:
# `skip_defaults=True` drops a `directed: true` the author *did* write
# (it equals the default), while `skip_defaults=False` writes one they
# did not. Both would answer the two declaration assertions with
# confident nonsense. `check_manifest` degrades them to a warning that
# says exactly this, which is the accurate report for a merged result.
report = check_manifest(
merged, profile=profile_name, subject=f"{left} + {right}"
)
for line in report.to_lines():
click.echo(line)
if dry_run:
_echo_naming(left_manifest, right_manifest, op, err=False)
lines = routing_table(routing)
if lines:
click.echo("routing:")
for line in lines:
click.echo(f" {line}")
click.echo("dry run: nothing written")
return
entry = None
if record_label is not None:
# Before `dump_manifest`, so the file on disk carries its own lineage --
# the whole point of provenance travelling with the artifact.
entry = _record(left_manifest, right_manifest, merged, op, store, record_label)
if output is not None:
dump_manifest(merged, output)
if entry is not None:
click.echo(f"commit: {entry.id}")
click.echo(f"stored: {append_entry(store, entry)}")