@click.command("merge")
@click.argument("left", type=click.Path(exists=True, dir_okay=False, path_type=Path))
@click.argument("right", 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, "
"canonical maps and identity alignments. 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",
# `fuse_right` is the pre-rename spelling of `union_right`; both are
# accepted here so a recorded command line keeps working.
type=click.Choice(["error", "prefix_right", "union_right", "fuse_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 by name (each shared or alike-spelled name becomes a 1-1 "
"equivalence into the left spelling); 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(
"--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,
right: Path,
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,
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)
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
payload: dict[str, Any] = load_mapping(op_path) if op_path is not None else {}
if name_conflict is not None:
payload["name_conflict"] = name_conflict
try:
_fold_canonical_maps(payload, canonical_map_paths)
# `op` is a Literal with a default, so a document carrying
# `op: merge_manifests` validates as written -- no key to strip.
op = MergeManifestsOp.model_validate(payload)
except ValueError as exc:
raise _MergeSetupError(f"{op_path}: invalid merge op -- {exc}") from exc
wants_preview = plot_path is not None or preview_json_path 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
)
def emit(outcome: MergeOutcome, subjects: tuple[str, ...] = ()) -> None:
if preview is None:
return
_write_preview(
preview.with_outcome(outcome, subjects=subjects),
plot_path=plot_path,
json_path=preview_json_path,
max_rows=max_rows,
)
try:
merged = merge_manifests(
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)
click.echo("completion:", err=True)
click.echo(
yaml.safe_dump(exc.completion.to_dict(), sort_keys=False).rstrip(),
err=True,
)
raise SystemExit(EXIT_REFUSED)
except (
AlignmentConflictError,
ClusterConflictError,
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)
raise SystemExit(EXIT_REFUSED)
emit(outcome_from_manifest(merged))
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 honest 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:
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)}")