From 4f41eaca0e4e001b95ecb47c66b432592bbcee66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Deuerling?= Date: Tue, 14 Jul 2026 09:55:18 +0200 Subject: [PATCH 1/4] Add apply/export commands to copy portable org state between instances Add a new `apply --source --to ` command that copies an organization's portable, declarative state from one Entropy Data instance to another by exporting the source into a staged directory and importing it into the target. - New `export dir ` enumerates a source into a local YAML tree (one directory per resource, one file per resource by id), mirroring the app's organization-export layout. - `import` gains an `import dir ` subcommand; `import zip` now extracts and reuses the same dir-import engine. Both gain --prune (reverse-dependency-order delete of resources absent from the import, confirmation-gated unless --yes) and --include/--exclude; `import dir` and `apply` also support --dry-run. - Fix `import` silently dropping sourcesystems from an org export. - export/import/apply now also copy certifications, classification-schemes and example-data. Writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. The experimental semantics API is not yet included. Resource ordering, identity fields and per-resource handling live in one shared module so export, import and apply cannot drift. --- CHANGELOG.md | 6 + README.md | 45 ++- src/entropy_data/cli.py | 21 +- src/entropy_data/commands/apply.py | 86 +++++ src/entropy_data/commands/import_export.py | 191 +++++----- src/entropy_data/resources.py | 110 ++++++ src/entropy_data/sync.py | 341 ++++++++++++++++++ tests/commands/test_import_export.py | 398 +++++++++++++++++++++ uv.lock | 2 +- 9 files changed, 1105 insertions(+), 95 deletions(-) create mode 100644 src/entropy_data/commands/apply.py create mode 100644 src/entropy_data/resources.py create mode 100644 src/entropy_data/sync.py create mode 100644 tests/commands/test_import_export.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d0cb200..b940ccb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,12 @@ ## [Unreleased] +- Add `entropy-data export dir ` to enumerate an organization's portable state into a local YAML tree (one directory per resource, one file per resource named by its id), mirroring the app's organization-export layout so the artifacts interchange. +- Extend `entropy-data import` with an `import dir ` subcommand alongside `import zip` (the zip form now extracts to a temp directory and reuses the same dir-import logic). Both gain `--prune` to delete target resources absent from the import (in reverse dependency order, guarded by a confirmation prompt unless `--yes`), plus `--include`/`--exclude` filters; `import dir` also supports `--dry-run`. +- Add `entropy-data apply --source --to ` to copy portable organization state from one connection to another by exporting the source into a staged directory and importing it into the target. Supports `--prune`, `--dry-run` (per-resource create/update/prune counts), `--include`/`--exclude`, and `--keep ` to retain the staged export. +- Fix `import` silently dropping `sourcesystems/` from an organization export — source systems are now imported in their correct dependency position (after policies, before assets). +- `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, and `example-data`. All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. The experimental `semantics` API is not yet included. + ## [0.3.19] - Fix `entropy-data datacontracts yaml ` requesting `/api/datacontracts/{id}.yaml`, which does not exist on the server, resulting in a 404. It now requests the documented `/api/datacontracts/{id}/datacontract.yaml`. diff --git a/README.md b/README.md index 6524f54..c422565 100644 --- a/README.md +++ b/README.md @@ -111,9 +111,52 @@ entropy-data [--version] [--connection NAME] [--output table|json|yaml] [--debug search query semantics namespaces ... | concepts ... | relationships ... | search usage list | submit | delete - import zip + export dir + import zip | dir + apply --to TARGET [--source SRC] [--prune] [--dry-run] [--include] [--exclude] [--keep DIR] ``` +## Copying organization state between instances + +`export`, `import` and `apply` move the portable declarative state of an organization +(teams, tags, definitions, policies, source systems, certifications, classification +schemes, assets, data contracts, data products, example data, access agreements) +between Entropy Data instances — for example to promote a test environment to prod. +Only state reachable through the public `/api/**` API and portable across instances is +copied (no secrets, telemetry, or environment-specific identity). Every write is an +idempotent PUT-by-id, so runs converge and are safe to repeat. + +```bash +# Export a source instance to a local YAML tree (one file per resource). +entropy-data -c source export dir ./state + +# Import that tree into a target instance. +entropy-data -c target import dir ./state + +# Copy directly from one connection to another (export -> stage -> import). +entropy-data apply --source source --to target + +# Preview the changes without writing anything. +entropy-data apply --source source --to target --dry-run + +# Mirror: also delete target resources that are absent from the source. +entropy-data apply --source source --to target --prune +``` + +Useful options: + +- `--include a,b` / `--exclude a,b` — restrict the resource set (names as listed above). +- `--prune` — after upserts, delete target resources absent from the source, in reverse + dependency order. Guarded by a confirmation prompt unless `--yes` is passed. +- `--dry-run` — print per-resource create/update/(prune) counts; no writes. +- `--keep DIR` (apply) — retain the staged export instead of using a temporary directory. + +Notes: + +- Team members are stripped on import (users are per-instance identities); the export + keeps them so the artifact is a faithful snapshot. +- The semantics (experimental) API is not yet part of the copy set. + ## Connection Management Connections are stored in `~/.entropy-data/config.toml`: diff --git a/src/entropy_data/cli.py b/src/entropy_data/cli.py index 7901520..18894ad 100644 --- a/src/entropy_data/cli.py +++ b/src/entropy_data/cli.py @@ -33,6 +33,16 @@ def get_client() -> EntropyDataClient: return EntropyDataClient(config) +def client_for_connection(name: str) -> EntropyDataClient: + """Create an API client for a specific named connection (for `apply --to/--source`). + + Resolves the connection by name only — the global --api-key/--host overrides + target the primary connection and must not bleed into a second endpoint. + """ + config = resolve_connection(connection_name=name) + return EntropyDataClient(config) + + def get_output_format() -> OutputFormat: return _output_format @@ -116,6 +126,7 @@ def main( # Register command groups from entropy_data.commands.access import access_app # noqa: E402 from entropy_data.commands.api_keys import api_keys_app # noqa: E402 +from entropy_data.commands.apply import apply_command # noqa: E402 from entropy_data.commands.assets import assets_app # noqa: E402 from entropy_data.commands.certifications import certifications_app # noqa: E402 from entropy_data.commands.classifications import classifications_app # noqa: E402 @@ -127,7 +138,7 @@ def main( from entropy_data.commands.definitions import definitions_app # noqa: E402 from entropy_data.commands.events import events_app # noqa: E402 from entropy_data.commands.example_data import example_data_app # noqa: E402 -from entropy_data.commands.import_export import import_app # noqa: E402 +from entropy_data.commands.import_export import export_app, import_app # noqa: E402 from entropy_data.commands.integrations import integrations_app # noqa: E402 from entropy_data.commands.lineage import lineage_app # noqa: E402 from entropy_data.commands.organization import organization_app # noqa: E402 @@ -164,8 +175,14 @@ def main( app.add_typer(settings_app, name="settings", help="Manage organization settings.") app.add_typer(events_app, name="events", help="Poll events.") app.add_typer(lineage_app, name="lineage", help="Manage lineage (OpenLineage events).") -app.add_typer(schemas_app, name="schemas", help="Get the JSON Schemas that data contracts (ODCS) and data products (ODPS) validate against.") +app.add_typer( + schemas_app, + name="schemas", + help="Get the JSON Schemas that data contracts (ODCS) and data products (ODPS) validate against.", +) app.add_typer(search_app, name="search", help="Search across resources.") app.add_typer(semantics_app, name="semantics", help="EXPERIMENTAL semantics API.") app.add_typer(usage_app, name="usage", help="Manage usage (OpenTelemetry traces).") app.add_typer(import_app, name="import", help="Import organization exports.") +app.add_typer(export_app, name="export", help="Export organization state to a local YAML tree.") +app.command(name="apply", help="Copy portable organization state from a source to a target connection.")(apply_command) diff --git a/src/entropy_data/commands/apply.py b/src/entropy_data/commands/apply.py new file mode 100644 index 0000000..972be98 --- /dev/null +++ b/src/entropy_data/commands/apply.py @@ -0,0 +1,86 @@ +"""`apply` — copy portable org state from a source instance to a target instance. + +Mechanism: export the source into a staged directory, then import that directory +into the target. Both halves reuse the ``export dir`` / ``import dir`` engine, so +the staged artifact is auditable and identical to a manual ``export`` + ``import``. +""" + +import shutil +import tempfile +from pathlib import Path +from typing import Annotated, Optional + +import typer + +from entropy_data.commands.import_export import _parse_csv, print_plan +from entropy_data.output import console, error_console +from entropy_data.resources import select_resources +from entropy_data.sync import export_dir, import_dir, plan_import + + +def apply_command( + to: Annotated[str, typer.Option("--to", help="Target named connection to apply state onto.")], + source: Annotated[ + Optional[str], + typer.Option("--source", help="Source named connection (defaults to the global -c/--connection)."), + ] = None, + prune: Annotated[bool, typer.Option("--prune", help="Delete target resources absent from the source.")] = False, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip the prune confirmation prompt.")] = False, + dry_run: Annotated[ + bool, typer.Option("--dry-run", help="Print planned create/update/(prune) counts; write nothing to the target.") + ] = False, + keep: Annotated[ + Optional[Path], typer.Option("--keep", help="Retain the staged export in this directory (else a temp dir).") + ] = None, + include: Annotated[Optional[str], typer.Option("--include", help="Comma-separated resources to include.")] = None, + exclude: Annotated[Optional[str], typer.Option("--exclude", help="Comma-separated resources to exclude.")] = None, +) -> None: + """Copy portable organization state from a source connection to a target connection.""" + from entropy_data.cli import client_for_connection, get_client, handle_error + + try: + resources = select_resources(_parse_csv(include), _parse_csv(exclude)) + source_client = client_for_connection(source) if source else get_client() + target_client = client_for_connection(to) + except Exception as e: + handle_error(e) + return + + if keep is not None: + staged = keep + staged.mkdir(parents=True, exist_ok=True) + cleanup = False + else: + staged = Path(tempfile.mkdtemp(prefix="entropy-data-apply-")) + cleanup = True + + try: + console.print(f"[bold]Exporting source into {staged}[/bold]") + export_result = export_dir(source_client, staged, resources) + console.print(f"\n[bold]Export:[/bold] {export_result.ok} exported, {export_result.fail} failed") + + if dry_run: + print_plan(plan_import(target_client, staged, resources, prune=prune)) + if export_result.fail > 0: + raise typer.Exit(1) + return + + if prune and not yes: + console.print("[yellow]--prune will DELETE target resources absent from the source.[/yellow]") + if not typer.confirm("Proceed with prune?"): + console.print("Aborted.") + raise typer.Exit(1) + + console.print("\n[bold]Applying to target[/bold]") + import_result = import_dir(target_client, staged, resources, prune=prune) + console.print( + f"\n[bold]Summary:[/bold] {import_result.ok} succeeded, {import_result.fail} failed " + f"({export_result.fail} export failures)" + ) + if import_result.fail > 0 or export_result.fail > 0: + raise typer.Exit(1) + finally: + if cleanup: + shutil.rmtree(staged, ignore_errors=True) + elif not dry_run: + error_console.print(f"Staged export retained at {staged}") diff --git a/src/entropy_data/commands/import_export.py b/src/entropy_data/commands/import_export.py index 2718b09..5bd683c 100644 --- a/src/entropy_data/commands/import_export.py +++ b/src/entropy_data/commands/import_export.py @@ -1,93 +1,52 @@ -"""Import command for organization exports.""" +"""Import and export commands for organization state. + +`export dir` enumerates a source instance into a local YAML tree; `import dir` +upserts such a tree into a target instance; `import zip` extracts an app-produced +org-export zip and imports it the same way. All share the engine in +``entropy_data.sync`` and the resource definitions in ``entropy_data.resources``. +""" import tempfile import zipfile from pathlib import Path -from typing import Annotated +from typing import Annotated, Optional import typer -import yaml -from entropy_data.client import ApiError from entropy_data.output import console, error_console +from entropy_data.resources import select_resources +from entropy_data.sync import export_dir, import_dir, plan_import import_app = typer.Typer(no_args_is_help=True) +export_app = typer.Typer(no_args_is_help=True) + + +def _parse_csv(value: str | None) -> list[str] | None: + if not value: + return None + return [item.strip() for item in value.split(",") if item.strip()] + -# Import order respects resource dependencies: -# teams (parent→child), tags (→teams), definitions (→teams), -# policies (independent), assets (→teams), datacontracts (→teams, assets), -# dataproducts (→teams, datacontracts, assets), access (→dataproducts) -RESOURCE_ORDER = [ - ("teams", "teams"), - ("tags", "tags"), - ("definitions", "definitions"), - ("policies", "policies"), - ("assets", "assets"), - ("datacontracts", "datacontracts"), - ("dataproducts", "dataproducts"), - ("access", "access"), -] - - -def _import_teams(teams_dir: Path, client) -> tuple[int, int]: - """Import teams in topological order (parents before children), stripping members.""" - teams = {} - for f in sorted(teams_dir.glob("*.yaml")): - data = yaml.safe_load(f.read_text()) - data["members"] = [] - teams[data["id"]] = {"data": data, "parent": data.get("parent")} - - imported: set[str] = set() - success_count = 0 - error_count = 0 - - while len(imported) < len(teams): - progress = False - for tid, t in teams.items(): - if tid in imported: - continue - if t["parent"] is None or t["parent"] in imported: - try: - client.put_resource("teams", tid, t["data"]) - console.print(f" [green]OK[/green] {tid}") - success_count += 1 - except ApiError as e: - error_console.print(f" [red]FAIL[/red] {tid}: {e}") - error_count += 1 - imported.add(tid) - progress = True - - if not progress: - remaining = set(teams.keys()) - imported - error_console.print(f" [red]ERROR: circular or broken parent references: {remaining}[/red]") - error_count += len(remaining) - break - - return success_count, error_count - - -def _import_simple(resource_dir: Path, api_path: str, client) -> tuple[int, int]: - """Import resources from a directory using PUT.""" - success_count = 0 - error_count = 0 - - for f in sorted(resource_dir.glob("*.yaml")): - data = yaml.safe_load(f.read_text()) - resource_id = data["id"] - try: - client.put_resource(api_path, resource_id, data) - console.print(f" [green]OK[/green] {resource_id}") - success_count += 1 - except ApiError as e: - error_console.print(f" [red]FAIL[/red] {resource_id}: {e}") - error_count += 1 - - return success_count, error_count +def print_plan(plan) -> None: + """Print a dry-run plan table of create/update/(prune) counts per resource.""" + if not plan.counts: + console.print("Nothing to do.") + return + console.print("\n[bold]Dry run — planned changes:[/bold]") + for name, counts in plan.counts.items(): + parts = [f"create={counts.create}", f"update={counts.update}"] + if counts.prune: + parts.append(f"prune={counts.prune}") + console.print(f" {name}: {', '.join(parts)}") @import_app.command("zip") def import_zip( file: Annotated[Path, typer.Argument(help="Path to the export zip file.")], + prune: Annotated[bool, typer.Option("--prune", help="Delete target resources absent from the import.")] = False, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip the prune confirmation prompt.")] = False, + include: Annotated[Optional[str], typer.Option("--include", help="Comma-separated resources to include.")] = None, + exclude: Annotated[Optional[str], typer.Option("--exclude", help="Comma-separated resources to exclude.")] = None, ) -> None: """Import an organization export zip file.""" from entropy_data.cli import get_client, handle_error @@ -101,35 +60,85 @@ def import_zip( raise typer.Exit(1) try: + resources = select_resources(_parse_csv(include), _parse_csv(exclude)) client = get_client() except Exception as e: handle_error(e) return with tempfile.TemporaryDirectory() as tmpdir: - export_dir = Path(tmpdir) + extracted = Path(tmpdir) console.print(f"Extracting {file}...") with zipfile.ZipFile(file) as zf: - zf.extractall(export_dir) + zf.extractall(extracted) + + _run_import(client, extracted, resources, prune, yes) - total_ok = 0 - total_fail = 0 - for directory, api_path in RESOURCE_ORDER: - resource_dir = export_dir / directory - if not resource_dir.is_dir(): - continue +@import_app.command("dir") +def import_dir_command( + path: Annotated[Path, typer.Argument(help="Directory holding the export YAML tree.")], + prune: Annotated[bool, typer.Option("--prune", help="Delete target resources absent from the import.")] = False, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip the prune confirmation prompt.")] = False, + dry_run: Annotated[ + bool, typer.Option("--dry-run", help="Print planned create/update/(prune) counts; write nothing.") + ] = False, + include: Annotated[Optional[str], typer.Option("--include", help="Comma-separated resources to include.")] = None, + exclude: Annotated[Optional[str], typer.Option("--exclude", help="Comma-separated resources to exclude.")] = None, +) -> None: + """Import an organization export directory tree.""" + from entropy_data.cli import get_client, handle_error + + if not path.is_dir(): + error_console.print(f"[red]Error: {path} is not a directory[/red]") + raise typer.Exit(1) - console.print(f"\n[bold]{api_path}[/bold]") + try: + resources = select_resources(_parse_csv(include), _parse_csv(exclude)) + client = get_client() + except Exception as e: + handle_error(e) + return - if directory == "teams": - ok, fail = _import_teams(resource_dir, client) - else: - ok, fail = _import_simple(resource_dir, api_path, client) + if dry_run: + print_plan(plan_import(client, path, resources, prune=prune)) + return - total_ok += ok - total_fail += fail + _run_import(client, path, resources, prune, yes) - console.print(f"\n[bold]Summary:[/bold] {total_ok} succeeded, {total_fail} failed") - if total_fail > 0: + +def _run_import(client, source: Path, resources, prune: bool, yes: bool) -> None: + """Shared import driver for both ``zip`` and ``dir``; prompts before pruning.""" + if prune and not yes: + console.print("[yellow]--prune will DELETE target resources absent from the import.[/yellow]") + if not typer.confirm("Proceed with prune?"): + console.print("Aborted.") raise typer.Exit(1) + + result = import_dir(client, source, resources, prune=prune) + console.print(f"\n[bold]Summary:[/bold] {result.ok} succeeded, {result.fail} failed") + if result.fail > 0: + raise typer.Exit(1) + + +@export_app.command("dir") +def export_dir_command( + path: Annotated[Path, typer.Argument(help="Destination directory for the export YAML tree.")], + include: Annotated[Optional[str], typer.Option("--include", help="Comma-separated resources to include.")] = None, + exclude: Annotated[Optional[str], typer.Option("--exclude", help="Comma-separated resources to exclude.")] = None, +) -> None: + """Export organization state to a local YAML tree (one file per resource).""" + from entropy_data.cli import get_client, handle_error + + try: + resources = select_resources(_parse_csv(include), _parse_csv(exclude)) + client = get_client() + except Exception as e: + handle_error(e) + return + + path.mkdir(parents=True, exist_ok=True) + result = export_dir(client, path, resources) + console.print(f"\n[bold]Summary:[/bold] {result.ok} exported, {result.fail} failed") + if result.fail > 0: + raise typer.Exit(1) diff --git a/src/entropy_data/resources.py b/src/entropy_data/resources.py new file mode 100644 index 0000000..1949d07 --- /dev/null +++ b/src/entropy_data/resources.py @@ -0,0 +1,110 @@ +"""Shared definition of the portable resource set and its dependency order. + +`export`, `import` and `apply` all consume :data:`RESOURCE_ORDER` so they cannot +drift: the same directory names, API paths, identity fields and per-resource +handling flags drive enumeration (export), upsert (import) and orchestration +(apply). + +The canonical dependency order is: + + teams -> tags -> definitions -> policies -> sourcesystems -> + certifications -> classification-schemes -> assets -> + datacontracts -> dataproducts -> example-data -> access + +Pruning walks this list in reverse so dependents are removed before their +dependencies. +""" + +from dataclasses import dataclass + +# Persistence bookkeeping that must never be sent back on a PUT. The public API +# does not accept these, and a GET should not surface them, but strip defensively. +AUDIT_FIELDS = ("createdAt", "createdBy", "updatedAt", "updatedBy", "created_at", "updated_at") + +# Characters that are invalid in filenames on common filesystems. Mirrors the +# app's OrganizationExportService.sanitizeFilename so the artifacts interchange. +_FILENAME_INVALID = '\\/:*?"<>|' + + +@dataclass(frozen=True) +class Resource: + """One copyable resource type. + + name directory name in the artifact tree and logical key for filters + api_path path segment for ``/api/{api_path}`` list/get/put/delete + id_field body field holding the stable identity (path id + filename) + paginated list follows ``Link: rel="next"`` (``p`` query param) + detail GET each item individually for the full body (list is partial) + strip_members drop ``members`` before PUT (teams: users are per-instance) + topo_sort_parents order writes so a parent is created before its children + tag_assignments after PUT, replay ``assignedTags`` via the sub-resource + endpoint ``PUT /assets/{id}/assigned-tags/{tagId}`` + """ + + name: str + api_path: str + id_field: str = "id" + paginated: bool = False + detail: bool = False + strip_members: bool = False + topo_sort_parents: bool = False + tag_assignments: bool = False + + +RESOURCE_ORDER: list[Resource] = [ + Resource("teams", "teams", paginated=True, strip_members=True, topo_sort_parents=True), + Resource("tags", "tags", paginated=True), + Resource("definitions", "definitions", paginated=True), + Resource("policies", "policies"), + Resource("sourcesystems", "sourcesystems", paginated=True), + Resource("certifications", "certifications"), + Resource("classification-schemes", "classification-schemes", id_field="externalId"), + # assets list omits assignedTags and returns a partial body, so fetch each one. + Resource("assets", "assets", paginated=True, detail=True, tag_assignments=True), + # datacontracts / dataproducts list endpoints return only top-level fields. + Resource("datacontracts", "datacontracts", detail=True), + Resource("dataproducts", "dataproducts", detail=True), + Resource("example-data", "example-data"), + Resource("access", "access", paginated=True), + # TODO: semantics (namespaces -> concepts -> relationships) is deliberately + # excluded. The experimental semantics API is nested (concepts and + # relationships live under a namespace path), which does not fit this flat + # one-directory-per-resource model without a bespoke multi-level walker. Add + # nested export/import support if/when semantics graduates from experimental. +] + +RESOURCE_BY_NAME: dict[str, Resource] = {r.name: r for r in RESOURCE_ORDER} + + +def sanitize_filename(resource_id: str) -> str: + """Make a resource id safe as a filename (ids may contain ``/``, ``:`` etc.).""" + return "".join("_" if c in _FILENAME_INVALID else c for c in resource_id) + + +def strip_audit_fields(body: dict) -> dict: + """Return a copy of ``body`` without top-level audit fields.""" + return {k: v for k, v in body.items() if k not in AUDIT_FIELDS} + + +def select_resources( + include: list[str] | None = None, + exclude: list[str] | None = None, +) -> list[Resource]: + """Filter :data:`RESOURCE_ORDER` by name, preserving dependency order. + + Raises ``ValueError`` naming any unknown resource so a typo in ``--include`` + or ``--exclude`` fails loudly instead of silently copying nothing. + """ + unknown = sorted({n for n in (include or []) + (exclude or []) if n not in RESOURCE_BY_NAME}) + if unknown: + known = ", ".join(r.name for r in RESOURCE_ORDER) + raise ValueError(f"Unknown resource(s): {', '.join(unknown)}. Known resources: {known}") + + result = RESOURCE_ORDER + if include: + included = set(include) + result = [r for r in result if r.name in included] + if exclude: + excluded = set(exclude) + result = [r for r in result if r.name not in excluded] + return result diff --git a/src/entropy_data/sync.py b/src/entropy_data/sync.py new file mode 100644 index 0000000..5c88861 --- /dev/null +++ b/src/entropy_data/sync.py @@ -0,0 +1,341 @@ +"""Export / import / prune engine for the portable resource set. + +All three user-facing commands (`export dir`, `import dir`/`import zip`, `apply`) +are thin wrappers over the functions here so they share one enumeration, upsert +and prune implementation and cannot drift. Everything is continue-on-error: a +single resource failure is logged and counted, never aborts the run. +""" + +from dataclasses import dataclass, field +from pathlib import Path + +import yaml + +from entropy_data.client import ( + REQUEST_TIMEOUT, + ApiError, + EntropyDataClient, + _raise_for_status, + _validate_resource_id, +) +from entropy_data.output import console, error_console +from entropy_data.resources import Resource, sanitize_filename, strip_audit_fields + + +@dataclass +class SyncResult: + """Tally of an export or import run.""" + + ok: int = 0 + fail: int = 0 + + def add(self, other: "SyncResult") -> None: + self.ok += other.ok + self.fail += other.fail + + +@dataclass +class PlanCounts: + """Per-resource dry-run plan.""" + + create: int = 0 + update: int = 0 + prune: int = 0 + + +# --- enumeration (source side) -------------------------------------------------- + + +def _list_all(client: EntropyDataClient, resource: Resource) -> list[dict]: + """Return every item of a resource, following ``Link: rel="next"`` if paginated.""" + if not resource.paginated: + items, _ = client.list_resources(resource.api_path) + return list(items) + + items: list[dict] = [] + page = 0 + while True: + batch, has_next = client.list_resources(resource.api_path, params={"p": page}) + items.extend(batch) + if not has_next: + break + page += 1 + return items + + +def _enumerate_ids(client: EntropyDataClient, resource: Resource) -> set[str]: + """Return the set of resource ids currently on the target (for prune / dry-run).""" + return {str(item[resource.id_field]) for item in _list_all(client, resource) if item.get(resource.id_field)} + + +# --- assigned-tags sub-resource (assets) --------------------------------------- + + +def _get_assigned_tags(client: EntropyDataClient, asset_id: str) -> list[str]: + _validate_resource_id(asset_id) + response = client.session.get( + f"{client.base_url}/api/assets/{asset_id}/assigned-tags", + timeout=REQUEST_TIMEOUT, + ) + _raise_for_status(response) + return response.json() + + +def _put_assigned_tag(client: EntropyDataClient, asset_id: str, tag_id: str) -> None: + _validate_resource_id(asset_id) + # tag_id may be hierarchical ("governance/PII"); the server validates it. + response = client.session.put( + f"{client.base_url}/api/assets/{asset_id}/assigned-tags/{tag_id}", + timeout=REQUEST_TIMEOUT, + ) + _raise_for_status(response) + + +def _apply_tag_assignments(client: EntropyDataClient, asset_id: str, body: dict) -> SyncResult: + """Replay an asset's ``assignedTags`` via the sub-resource endpoint. + + The asset PUT ignores ``assignedTags``, so tags are assigned separately. + Idempotent: only tags not already present are added. + """ + result = SyncResult() + desired = body.get("assignedTags") or [] + if not desired: + return result + try: + existing = set(_get_assigned_tags(client, asset_id)) + except ApiError: + existing = set() + for tag in desired: + if tag in existing: + continue + try: + _put_assigned_tag(client, asset_id, tag) + console.print(f" [green]OK[/green] tag {tag}") + result.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] tag {tag}: {e}") + result.fail += 1 + return result + + +# --- export -------------------------------------------------------------------- + + +def export_dir(client: EntropyDataClient, dest: Path, resources: list[Resource]) -> SyncResult: + """Enumerate the source and write one YAML file per resource under ``dest``.""" + total = SyncResult() + for resource in resources: + console.print(f"\n[bold]{resource.name}[/bold]") + try: + items = _list_all(client, resource) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + total.fail += 1 + continue + + if not items: + console.print(" (none)") + continue + + target_dir = dest / resource.name + target_dir.mkdir(parents=True, exist_ok=True) + + for item in items: + rid = item.get(resource.id_field) + if rid is None: + error_console.print(f" [red]FAIL[/red] item without '{resource.id_field}'") + total.fail += 1 + continue + rid = str(rid) + try: + body = client.get_resource(resource.api_path, rid) if resource.detail else item + body = strip_audit_fields(body) + path = target_dir / f"{sanitize_filename(rid)}.yaml" + path.write_text(yaml.safe_dump(body, sort_keys=False, allow_unicode=True, width=4096)) + console.print(f" [green]OK[/green] {rid}") + total.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {rid}: {e}") + total.fail += 1 + return total + + +# --- import -------------------------------------------------------------------- + + +def _load_dir(resource_dir: Path, resource: Resource) -> list[tuple[str, dict]]: + """Read every ``*.yaml`` file in a resource directory into (id, body) pairs.""" + entries: list[tuple[str, dict]] = [] + for f in sorted(resource_dir.glob("*.yaml")): + data = yaml.safe_load(f.read_text()) + if not isinstance(data, dict): + continue + rid = data.get(resource.id_field) + if rid is None: + continue + entries.append((str(rid), strip_audit_fields(data))) + return entries + + +def _upsert_teams(client: EntropyDataClient, resource: Resource, entries: list[tuple[str, dict]]) -> SyncResult: + """Upsert teams parents-first, stripping members.""" + result = SyncResult() + teams = {rid: {"data": {**body, "members": []}, "parent": body.get("parent")} for rid, body in entries} + imported: set[str] = set() + + while len(imported) < len(teams): + progress = False + for tid, t in teams.items(): + if tid in imported: + continue + if t["parent"] is None or t["parent"] in imported: + try: + client.put_resource(resource.api_path, tid, t["data"]) + console.print(f" [green]OK[/green] {tid}") + result.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {tid}: {e}") + result.fail += 1 + imported.add(tid) + progress = True + if not progress: + remaining = set(teams) - imported + error_console.print(f" [red]ERROR: circular or broken parent references: {remaining}[/red]") + result.fail += len(remaining) + break + return result + + +def _upsert(client: EntropyDataClient, resource: Resource, entries: list[tuple[str, dict]]) -> SyncResult: + """Upsert one resource type via PUT-by-id (idempotent), continue-on-error.""" + if resource.topo_sort_parents: + return _upsert_teams(client, resource, entries) + + result = SyncResult() + for rid, body in entries: + payload = {**body, "members": []} if resource.strip_members else body + try: + client.put_resource(resource.api_path, rid, payload) + console.print(f" [green]OK[/green] {rid}") + result.ok += 1 + if resource.tag_assignments: + result.add(_apply_tag_assignments(client, rid, payload)) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {rid}: {e}") + result.fail += 1 + return result + + +def _prune( + client: EntropyDataClient, + resources: list[Resource], + imported_ids: dict[str, set[str]], +) -> SyncResult: + """Delete target resources absent from the import set, in reverse dependency order.""" + result = SyncResult() + for resource in reversed(resources): + keep = imported_ids.get(resource.name, set()) + try: + existing = _list_all(client, resource) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + result.fail += 1 + continue + + to_delete = [item for item in existing if str(item.get(resource.id_field)) not in keep] + if not to_delete: + continue + + console.print(f"\n[bold]prune {resource.name}[/bold]") + ordered = _order_for_deletion(resource, to_delete) + for item in ordered: + rid = str(item[resource.id_field]) + try: + client.delete_resource(resource.api_path, rid) + console.print(f" [green]DEL[/green] {rid}") + result.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {rid}: {e}") + result.fail += 1 + return result + + +def _order_for_deletion(resource: Resource, items: list[dict]) -> list[dict]: + """For hierarchical resources, delete children before parents; else keep order.""" + if not resource.topo_sort_parents: + return items + + remaining = {str(item[resource.id_field]): item for item in items} + ordered: list[dict] = [] + while remaining: + # A team is a leaf here if no other team still queued names it as parent. + parents_in_play = {item.get("parent") for item in remaining.values()} + leaves = [rid for rid in remaining if rid not in parents_in_play] + if not leaves: # cycle or broken refs: fall back to whatever is left + leaves = list(remaining) + for rid in leaves: + ordered.append(remaining.pop(rid)) + return ordered + + +@dataclass +class ImportPlan: + """Result of a dry-run: per-resource create/update/prune counts.""" + + counts: dict[str, PlanCounts] = field(default_factory=dict) + + +def plan_import( + client: EntropyDataClient, + source: Path, + resources: list[Resource], + prune: bool = False, +) -> ImportPlan: + """Compute create/update/(prune) counts without writing anything.""" + plan = ImportPlan() + for resource in resources: + resource_dir = source / resource.name + entries = _load_dir(resource_dir, resource) if resource_dir.is_dir() else [] + imported_ids = {rid for rid, _ in entries} + + try: + existing_ids = _enumerate_ids(client, resource) + except ApiError as e: + error_console.print(f"[red]FAIL[/red] list {resource.name}: {e}") + existing_ids = set() + + counts = PlanCounts( + create=len(imported_ids - existing_ids), + update=len(imported_ids & existing_ids), + prune=len(existing_ids - imported_ids) if prune else 0, + ) + if counts.create or counts.update or counts.prune: + plan.counts[resource.name] = counts + return plan + + +def import_dir( + client: EntropyDataClient, + source: Path, + resources: list[Resource], + prune: bool = False, +) -> SyncResult: + """Upsert every resource under ``source`` in dependency order, optionally pruning.""" + total = SyncResult() + imported_ids: dict[str, set[str]] = {} + + for resource in resources: + resource_dir = source / resource.name + if not resource_dir.is_dir(): + continue + entries = _load_dir(resource_dir, resource) + imported_ids[resource.name] = {rid for rid, _ in entries} + if not entries: + continue + console.print(f"\n[bold]{resource.name}[/bold]") + total.add(_upsert(client, resource, entries)) + + if prune: + total.add(_prune(client, resources, imported_ids)) + + return total diff --git a/tests/commands/test_import_export.py b/tests/commands/test_import_export.py new file mode 100644 index 0000000..ea2b940 --- /dev/null +++ b/tests/commands/test_import_export.py @@ -0,0 +1,398 @@ +"""Tests for export / import dir / apply commands and the shared sync engine.""" + +import responses +import yaml +from typer.testing import CliRunner + +import entropy_data.config as cfg +from entropy_data.cli import app +from entropy_data.resources import RESOURCE_ORDER, select_resources + +runner = CliRunner() +BASE_URL = "https://api.entropy-data.com" +TARGET_URL = "https://target.example.com" + + +def _config_with_two_connections(tmp_path, monkeypatch): + """Write a config.toml with a source (default) and a target connection.""" + config_file = tmp_path / "config.toml" + monkeypatch.setattr(cfg, "CONFIG_FILE", config_file) + cfg.save_config( + { + "default_connection_name": "source", + "connections": { + "source": {"api_key": "src-key", "host": BASE_URL}, + "target": {"api_key": "tgt-key", "host": TARGET_URL}, + }, + } + ) + + +# --- resource ordering ---------------------------------------------------------- + + +def test_sourcesystems_is_included_in_order(): + names = [r.name for r in RESOURCE_ORDER] + assert "sourcesystems" in names + # Fixes the drop bug: sourcesystems sits after policies and before assets. + assert names.index("policies") < names.index("sourcesystems") < names.index("assets") + + +def test_new_resources_present_in_order(): + names = [r.name for r in RESOURCE_ORDER] + for expected in ("certifications", "classification-schemes", "example-data"): + assert expected in names + # example-data comes after dataproducts (its parent). + assert names.index("dataproducts") < names.index("example-data") + # canonical full order + assert names == [ + "teams", + "tags", + "definitions", + "policies", + "sourcesystems", + "certifications", + "classification-schemes", + "assets", + "datacontracts", + "dataproducts", + "example-data", + "access", + ] + + +def test_semantics_excluded_from_default_order(): + assert "semantics" not in [r.name for r in RESOURCE_ORDER] + + +def test_select_resources_include_exclude(): + only = select_resources(include=["teams", "tags"]) + assert [r.name for r in only] == ["teams", "tags"] + + without = select_resources(exclude=["access"]) + assert "access" not in [r.name for r in without] + + +def test_select_resources_unknown_raises(): + import pytest + + with pytest.raises(ValueError): + select_resources(include=["nope"]) + + +# --- export --------------------------------------------------------------------- + + +@responses.activate +def test_export_writes_files(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + # Only teams and certifications have content; everything else lists empty. + responses.add( + responses.GET, + f"{BASE_URL}/api/teams", + json=[{"id": "team-a", "name": "A", "members": [{"emailAddress": "x@y.z"}]}], + status=200, + ) + responses.add( + responses.GET, + f"{BASE_URL}/api/certifications", + json=[{"id": "gold", "name": "Gold", "createdAt": "2020-01-01"}], + status=200, + ) + for r in RESOURCE_ORDER: + if r.name in ("teams", "certifications"): + continue + responses.add(responses.GET, f"{BASE_URL}/api/{r.api_path}", json=[], status=200) + + dest = tmp_path / "export" + result = runner.invoke(app, ["export", "dir", str(dest)]) + assert result.exit_code == 0, result.output + + team_file = dest / "teams" / "team-a.yaml" + assert team_file.is_file() + # Export keeps members (faithful round-trip); import strips them. + assert yaml.safe_load(team_file.read_text())["members"] == [{"emailAddress": "x@y.z"}] + + cert_file = dest / "certifications" / "gold.yaml" + body = yaml.safe_load(cert_file.read_text()) + assert body["id"] == "gold" + # Audit fields stripped. + assert "createdAt" not in body + + +@responses.activate +def test_export_detail_fetches_each_item(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + # dataproducts list returns a partial body; the full body comes from GET /{id}. + responses.add(responses.GET, f"{BASE_URL}/api/dataproducts", json=[{"id": "dp-1"}], status=200) + responses.add( + responses.GET, + f"{BASE_URL}/api/dataproducts/dp-1", + json={"id": "dp-1", "info": {"title": "Full body"}}, + status=200, + ) + for r in RESOURCE_ORDER: + if r.name == "dataproducts": + continue + responses.add(responses.GET, f"{BASE_URL}/api/{r.api_path}", json=[], status=200) + + dest = tmp_path / "export" + result = runner.invoke(app, ["export", "dir", str(dest), "--include", "dataproducts"]) + assert result.exit_code == 0, result.output + body = yaml.safe_load((dest / "dataproducts" / "dp-1.yaml").read_text()) + assert body["info"]["title"] == "Full body" + + +# --- import dir ----------------------------------------------------------------- + + +def _write_tree(root, tree): + for dirname, files in tree.items(): + d = root / dirname + d.mkdir(parents=True, exist_ok=True) + for name, body in files.items(): + (d / f"{name}.yaml").write_text(yaml.safe_dump(body)) + + +@responses.activate +def test_import_dir_upserts_in_order(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree( + src, + { + "teams": { + "parent-team": {"id": "parent-team", "name": "Parent", "members": [{"emailAddress": "a@b.c"}]}, + "child-team": {"id": "child-team", "name": "Child", "parent": "parent-team"}, + }, + "certifications": {"gold": {"id": "gold", "name": "Gold"}}, + }, + ) + + puts = [] + + def _record(request): + puts.append(request.url) + return (200, {}, "") + + responses.add_callback(responses.PUT, f"{BASE_URL}/api/teams/parent-team", callback=_record) + responses.add_callback(responses.PUT, f"{BASE_URL}/api/teams/child-team", callback=_record) + responses.add_callback(responses.PUT, f"{BASE_URL}/api/certifications/gold", callback=_record) + + result = runner.invoke(app, ["import", "dir", str(src)]) + assert result.exit_code == 0, result.output + # Parent team upserted before its child. + assert puts.index(f"{BASE_URL}/api/teams/parent-team") < puts.index(f"{BASE_URL}/api/teams/child-team") + assert f"{BASE_URL}/api/certifications/gold" in puts + + +@responses.activate +def test_import_dir_strips_team_members(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree(src, {"teams": {"t1": {"id": "t1", "name": "T", "members": [{"emailAddress": "a@b.c"}]}}}) + + captured = {} + + def _capture(request): + import json as _json + + captured["body"] = _json.loads(request.body) + return (200, {}, "") + + responses.add_callback(responses.PUT, f"{BASE_URL}/api/teams/t1", callback=_capture) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "teams"]) + assert result.exit_code == 0, result.output + assert captured["body"]["members"] == [] + + +@responses.activate +def test_import_dir_assigns_asset_tags(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree( + src, + {"assets": {"asset-1": {"id": "asset-1", "info": {"name": "A"}, "assignedTags": ["pii", "governance/PII"]}}}, + ) + + responses.add(responses.PUT, f"{BASE_URL}/api/assets/asset-1", status=200) + # One tag already assigned -> only the missing one is added (idempotent). + responses.add(responses.GET, f"{BASE_URL}/api/assets/asset-1/assigned-tags", json=["pii"], status=200) + tag_puts = [] + + def _record_tag(request): + tag_puts.append(request.url) + return (200, {}, "") + + responses.add_callback( + responses.PUT, f"{BASE_URL}/api/assets/asset-1/assigned-tags/governance/PII", callback=_record_tag + ) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "assets"]) + assert result.exit_code == 0, result.output + assert tag_puts == [f"{BASE_URL}/api/assets/asset-1/assigned-tags/governance/PII"] + + +@responses.activate +def test_import_dir_dry_run_no_writes(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree( + src, + {"certifications": {"gold": {"id": "gold", "name": "Gold"}, "silver": {"id": "silver", "name": "Silver"}}}, + ) + + # Target already has "gold" -> gold is an update, silver is a create. + responses.add(responses.GET, f"{BASE_URL}/api/certifications", json=[{"id": "gold"}], status=200) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "certifications", "--dry-run"]) + assert result.exit_code == 0, result.output + assert "create=1" in result.output + assert "update=1" in result.output + # No PUT was registered; a write would have raised ConnectionError. + + +# --- prune ---------------------------------------------------------------------- + + +@responses.activate +def test_import_dir_prune_deletes_absent(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree(src, {"certifications": {"gold": {"id": "gold", "name": "Gold"}}}) + + responses.add(responses.PUT, f"{BASE_URL}/api/certifications/gold", status=200) + # Target has gold (kept) and bronze (absent from import -> pruned). + responses.add( + responses.GET, + f"{BASE_URL}/api/certifications", + json=[{"id": "gold"}, {"id": "bronze"}], + status=200, + ) + deletes = [] + + def _record_delete(request): + deletes.append(request.url) + return (200, {}, "") + + responses.add_callback(responses.DELETE, f"{BASE_URL}/api/certifications/bronze", callback=_record_delete) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "certifications", "--prune", "--yes"]) + assert result.exit_code == 0, result.output + # Only the absent resource is deleted. + assert deletes == [f"{BASE_URL}/api/certifications/bronze"] + + +@responses.activate +def test_import_dir_prune_deletes_teams_children_first(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + # Import is empty for teams -> both existing teams get pruned, child before parent. + (src / "teams").mkdir(parents=True) + + responses.add( + responses.GET, + f"{BASE_URL}/api/teams", + json=[{"id": "parent", "parent": None}, {"id": "child", "parent": "parent"}], + status=200, + ) + deletes = [] + + def _record(request): + deletes.append(request.url.rsplit("/", 1)[-1]) + return (200, {}, "") + + responses.add_callback(responses.DELETE, f"{BASE_URL}/api/teams/parent", callback=_record) + responses.add_callback(responses.DELETE, f"{BASE_URL}/api/teams/child", callback=_record) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "teams", "--prune", "--yes"]) + assert result.exit_code == 0, result.output + assert deletes == ["child", "parent"] + + +@responses.activate +def test_import_dir_prune_prompt_abort(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + _write_tree(src, {"certifications": {"gold": {"id": "gold", "name": "Gold"}}}) + + # Decline the confirmation prompt -> abort, no HTTP calls. + result = runner.invoke(app, ["import", "dir", str(src), "--include", "certifications", "--prune"], input="n\n") + assert result.exit_code == 1 + assert "Aborted" in result.output + + +# --- apply ---------------------------------------------------------------------- + + +@responses.activate +def test_apply_orchestrates_export_import(monkeypatch, tmp_path): + _config_with_two_connections(tmp_path, monkeypatch) + + # Source export: one certification, everything else empty. + responses.add(responses.GET, f"{BASE_URL}/api/certifications", json=[{"id": "gold", "name": "Gold"}], status=200) + for r in RESOURCE_ORDER: + if r.name == "certifications": + continue + responses.add(responses.GET, f"{BASE_URL}/api/{r.api_path}", json=[], status=200) + + # Target import: capture the PUT. + target_puts = [] + + def _record(request): + target_puts.append(request.url) + return (200, {}, "") + + responses.add_callback(responses.PUT, f"{TARGET_URL}/api/certifications/gold", callback=_record) + + keep = tmp_path / "staged" + result = runner.invoke(app, ["apply", "--to", "target", "--keep", str(keep)]) + assert result.exit_code == 0, result.output + assert target_puts == [f"{TARGET_URL}/api/certifications/gold"] + # Staged artifact retained. + assert (keep / "certifications" / "gold.yaml").is_file() + + +@responses.activate +def test_apply_dry_run_no_target_writes(monkeypatch, tmp_path): + _config_with_two_connections(tmp_path, monkeypatch) + + responses.add(responses.GET, f"{BASE_URL}/api/certifications", json=[{"id": "gold", "name": "Gold"}], status=200) + for r in RESOURCE_ORDER: + if r.name == "certifications": + continue + responses.add(responses.GET, f"{BASE_URL}/api/{r.api_path}", json=[], status=200) + + # Target enumeration for the plan (read-only); no certifications yet -> create. + for r in RESOURCE_ORDER: + responses.add(responses.GET, f"{TARGET_URL}/api/{r.api_path}", json=[], status=200) + + result = runner.invoke(app, ["apply", "--to", "target", "--dry-run"]) + assert result.exit_code == 0, result.output + assert "create=1" in result.output + + +def test_import_help_lists_dir_and_zip(): + result = runner.invoke(app, ["import", "--help"]) + assert result.exit_code == 0 + assert "zip" in result.output + assert "dir" in result.output diff --git a/uv.lock b/uv.lock index d9e8579..a763901 100644 --- a/uv.lock +++ b/uv.lock @@ -143,7 +143,7 @@ wheels = [ [[package]] name = "entropy-data" -version = "0.3.16" +version = "0.3.19" source = { editable = "." } dependencies = [ { name = "pydantic" }, From c78a2b68bcc321e8f0334187ad62daa013877e8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Deuerling?= Date: Tue, 14 Jul 2026 10:16:52 +0200 Subject: [PATCH 2/4] Add semantics to the copy set via nested resource support Extend the export/import/apply engine to handle nested (parent- parameterized) resources and include the semantics graph: semantic-namespaces (flat), and semantic-concepts / semantic- relationships nested per namespace. A nested Resource carries a parent name and a {parent} api_path template. Export lists the parent, then children per parent, writing //.yaml; import upserts namespace-first; prune enumerates parents and removes absent children per namespace (in reverse dependency order). Flat behavior is unchanged. --- CHANGELOG.md | 2 +- README.md | 7 +- src/entropy_data/resources.py | 41 +++- src/entropy_data/sync.py | 304 ++++++++++++++++++--------- tests/commands/test_import_export.py | 111 +++++++++- 5 files changed, 359 insertions(+), 106 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b940ccb..3c99f68 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,7 @@ - Extend `entropy-data import` with an `import dir ` subcommand alongside `import zip` (the zip form now extracts to a temp directory and reuses the same dir-import logic). Both gain `--prune` to delete target resources absent from the import (in reverse dependency order, guarded by a confirmation prompt unless `--yes`), plus `--include`/`--exclude` filters; `import dir` also supports `--dry-run`. - Add `entropy-data apply --source --to ` to copy portable organization state from one connection to another by exporting the source into a staged directory and importing it into the target. Supports `--prune`, `--dry-run` (per-resource create/update/prune counts), `--include`/`--exclude`, and `--keep ` to retain the staged export. - Fix `import` silently dropping `sourcesystems/` from an organization export — source systems are now imported in their correct dependency position (after policies, before assets). -- `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, and `example-data`. All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. The experimental `semantics` API is not yet included. +- `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, `example-data`, and the `semantics` graph (namespaces, and concepts/relationships nested per namespace). All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. Nested semantics resources are exported to `//.yaml`, imported namespace-first, and pruned per namespace. ## [0.3.19] diff --git a/README.md b/README.md index c422565..950ff12 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,8 @@ entropy-data [--version] [--connection NAME] [--output table|json|yaml] [--debug `export`, `import` and `apply` move the portable declarative state of an organization (teams, tags, definitions, policies, source systems, certifications, classification -schemes, assets, data contracts, data products, example data, access agreements) -between Entropy Data instances — for example to promote a test environment to prod. +schemes, assets, data contracts, data products, example data, access agreements, and the +semantics graph) between Entropy Data instances — for example to promote a test environment to prod. Only state reachable through the public `/api/**` API and portable across instances is copied (no secrets, telemetry, or environment-specific identity). Every write is an idempotent PUT-by-id, so runs converge and are safe to repeat. @@ -155,7 +155,8 @@ Notes: - Team members are stripped on import (users are per-instance identities); the export keeps them so the artifact is a faithful snapshot. -- The semantics (experimental) API is not yet part of the copy set. +- The semantics graph is nested: namespaces are copied first, then concepts and + relationships per namespace. Its artifact lives at `semantic-//.yaml`. ## Connection Management diff --git a/src/entropy_data/resources.py b/src/entropy_data/resources.py index 1949d07..1af0918 100644 --- a/src/entropy_data/resources.py +++ b/src/entropy_data/resources.py @@ -9,10 +9,17 @@ teams -> tags -> definitions -> policies -> sourcesystems -> certifications -> classification-schemes -> assets -> - datacontracts -> dataproducts -> example-data -> access + datacontracts -> dataproducts -> example-data -> access -> + semantic-namespaces -> semantic-concepts -> semantic-relationships Pruning walks this list in reverse so dependents are removed before their dependencies. + +Most resources are flat: ``/api/{api_path}`` lists the whole organization and +``/api/{api_path}/{id}`` addresses one. A nested resource (``parent`` set) lives +under a parent's path — its ``api_path`` is a ``{parent}`` template expanded per +parent id, and it is enumerated by listing the parent, then listing children for +each parent. Its artifact layout is ``//.yaml``. """ from dataclasses import dataclass @@ -39,6 +46,8 @@ class Resource: topo_sort_parents order writes so a parent is created before its children tag_assignments after PUT, replay ``assignedTags`` via the sub-resource endpoint ``PUT /assets/{id}/assigned-tags/{tagId}`` + parent name of the parent resource; when set, ``api_path`` is a + ``{parent}`` template expanded per parent id (nested resource) """ name: str @@ -49,6 +58,15 @@ class Resource: strip_members: bool = False topo_sort_parents: bool = False tag_assignments: bool = False + parent: str | None = None + + def path_for(self, parent_id: str | None = None) -> str: + """Resolve ``api_path`` for a given parent id (identity for flat resources).""" + if self.parent is None: + return self.api_path + if parent_id is None: + raise ValueError(f"Resource '{self.name}' is nested under '{self.parent}' and needs a parent id.") + return self.api_path.format(parent=parent_id) RESOURCE_ORDER: list[Resource] = [ @@ -66,11 +84,22 @@ class Resource: Resource("dataproducts", "dataproducts", detail=True), Resource("example-data", "example-data"), Resource("access", "access", paginated=True), - # TODO: semantics (namespaces -> concepts -> relationships) is deliberately - # excluded. The experimental semantics API is nested (concepts and - # relationships live under a namespace path), which does not fit this flat - # one-directory-per-resource model without a bespoke multi-level walker. Add - # nested export/import support if/when semantics graduates from experimental. + # Semantics is nested: concepts and relationships live under a namespace path + # and are enumerated per namespace. They carry their identity in `id` (which may + # contain "/"); the list endpoints may return partial bodies, so fetch each. + Resource("semantic-namespaces", "semantics/experimental/namespaces", id_field="namespace"), + Resource( + "semantic-concepts", + "semantics/experimental/namespaces/{parent}/concepts", + parent="semantic-namespaces", + detail=True, + ), + Resource( + "semantic-relationships", + "semantics/experimental/namespaces/{parent}/relationships", + parent="semantic-namespaces", + detail=True, + ), ] RESOURCE_BY_NAME: dict[str, Resource] = {r.name: r for r in RESOURCE_ORDER} diff --git a/src/entropy_data/sync.py b/src/entropy_data/sync.py index 5c88861..d616b35 100644 --- a/src/entropy_data/sync.py +++ b/src/entropy_data/sync.py @@ -4,6 +4,11 @@ are thin wrappers over the functions here so they share one enumeration, upsert and prune implementation and cannot drift. Everything is continue-on-error: a single resource failure is logged and counted, never aborts the run. + +Flat resources are `/api/{api_path}` (list-all) + `/api/{api_path}/{id}`. A nested +resource (``Resource.parent`` set) is enumerated by first listing its parent, then +listing children under each parent's expanded path; its artifact lives at +``//.yaml``. """ from dataclasses import dataclass, field @@ -19,7 +24,7 @@ _validate_resource_id, ) from entropy_data.output import console, error_console -from entropy_data.resources import Resource, sanitize_filename, strip_audit_fields +from entropy_data.resources import RESOURCE_BY_NAME, Resource, sanitize_filename, strip_audit_fields @dataclass @@ -46,16 +51,20 @@ class PlanCounts: # --- enumeration (source side) -------------------------------------------------- -def _list_all(client: EntropyDataClient, resource: Resource) -> list[dict]: - """Return every item of a resource, following ``Link: rel="next"`` if paginated.""" +def _list_all(client: EntropyDataClient, resource: Resource, parent_id: str | None = None) -> list[dict]: + """Return every item of a resource, following ``Link: rel="next"`` if paginated. + + ``parent_id`` selects the parent for a nested resource; it is ignored for flat ones. + """ + api_path = resource.path_for(parent_id) if not resource.paginated: - items, _ = client.list_resources(resource.api_path) + items, _ = client.list_resources(api_path) return list(items) items: list[dict] = [] page = 0 while True: - batch, has_next = client.list_resources(resource.api_path, params={"p": page}) + batch, has_next = client.list_resources(api_path, params={"p": page}) items.extend(batch) if not has_next: break @@ -63,9 +72,17 @@ def _list_all(client: EntropyDataClient, resource: Resource) -> list[dict]: return items -def _enumerate_ids(client: EntropyDataClient, resource: Resource) -> set[str]: +def _enumerate_ids(client: EntropyDataClient, resource: Resource, parent_id: str | None = None) -> set[str]: """Return the set of resource ids currently on the target (for prune / dry-run).""" - return {str(item[resource.id_field]) for item in _list_all(client, resource) if item.get(resource.id_field)} + return { + str(item[resource.id_field]) for item in _list_all(client, resource, parent_id) if item.get(resource.id_field) + } + + +def _parent_ids(client: EntropyDataClient, resource: Resource) -> list[str]: + """List the ids of a nested resource's parent (e.g. namespaces for concepts).""" + parent = RESOURCE_BY_NAME[resource.parent] + return [str(item[parent.id_field]) for item in _list_all(client, parent) if item.get(parent.id_field)] # --- assigned-tags sub-resource (assets) --------------------------------------- @@ -121,42 +138,59 @@ def _apply_tag_assignments(client: EntropyDataClient, asset_id: str, body: dict) # --- export -------------------------------------------------------------------- +def _export_items(client: EntropyDataClient, resource: Resource, parent_id: str | None, target_dir: Path) -> SyncResult: + """Enumerate one (flat resource) or one parent's children (nested) into ``target_dir``.""" + total = SyncResult() + api_path = resource.path_for(parent_id) + try: + items = _list_all(client, resource, parent_id) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + total.fail += 1 + return total + + if not items: + return total + + target_dir.mkdir(parents=True, exist_ok=True) + for item in items: + rid = item.get(resource.id_field) + if rid is None: + error_console.print(f" [red]FAIL[/red] item without '{resource.id_field}'") + total.fail += 1 + continue + rid = str(rid) + try: + body = client.get_resource(api_path, rid) if resource.detail else item + body = strip_audit_fields(body) + path = target_dir / f"{sanitize_filename(rid)}.yaml" + path.write_text(yaml.safe_dump(body, sort_keys=False, allow_unicode=True, width=4096)) + label = f"{parent_id}/{rid}" if parent_id else rid + console.print(f" [green]OK[/green] {label}") + total.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {rid}: {e}") + total.fail += 1 + return total + + def export_dir(client: EntropyDataClient, dest: Path, resources: list[Resource]) -> SyncResult: """Enumerate the source and write one YAML file per resource under ``dest``.""" total = SyncResult() for resource in resources: console.print(f"\n[bold]{resource.name}[/bold]") + if resource.parent is None: + total.add(_export_items(client, resource, None, dest / resource.name)) + continue + try: - items = _list_all(client, resource) + parent_ids = _parent_ids(client, resource) except ApiError as e: - error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + error_console.print(f" [red]FAIL[/red] list parents for {resource.name}: {e}") total.fail += 1 continue - - if not items: - console.print(" (none)") - continue - - target_dir = dest / resource.name - target_dir.mkdir(parents=True, exist_ok=True) - - for item in items: - rid = item.get(resource.id_field) - if rid is None: - error_console.print(f" [red]FAIL[/red] item without '{resource.id_field}'") - total.fail += 1 - continue - rid = str(rid) - try: - body = client.get_resource(resource.api_path, rid) if resource.detail else item - body = strip_audit_fields(body) - path = target_dir / f"{sanitize_filename(rid)}.yaml" - path.write_text(yaml.safe_dump(body, sort_keys=False, allow_unicode=True, width=4096)) - console.print(f" [green]OK[/green] {rid}") - total.ok += 1 - except ApiError as e: - error_console.print(f" [red]FAIL[/red] {rid}: {e}") - total.fail += 1 + for pid in parent_ids: + total.add(_export_items(client, resource, pid, dest / resource.name / sanitize_filename(pid))) return total @@ -177,7 +211,9 @@ def _load_dir(resource_dir: Path, resource: Resource) -> list[tuple[str, dict]]: return entries -def _upsert_teams(client: EntropyDataClient, resource: Resource, entries: list[tuple[str, dict]]) -> SyncResult: +def _upsert_teams( + client: EntropyDataClient, resource: Resource, api_path: str, entries: list[tuple[str, dict]] +) -> SyncResult: """Upsert teams parents-first, stripping members.""" result = SyncResult() teams = {rid: {"data": {**body, "members": []}, "parent": body.get("parent")} for rid, body in entries} @@ -190,7 +226,7 @@ def _upsert_teams(client: EntropyDataClient, resource: Resource, entries: list[t continue if t["parent"] is None or t["parent"] in imported: try: - client.put_resource(resource.api_path, tid, t["data"]) + client.put_resource(api_path, tid, t["data"]) console.print(f" [green]OK[/green] {tid}") result.ok += 1 except ApiError as e: @@ -206,57 +242,29 @@ def _upsert_teams(client: EntropyDataClient, resource: Resource, entries: list[t return result -def _upsert(client: EntropyDataClient, resource: Resource, entries: list[tuple[str, dict]]) -> SyncResult: +def _upsert( + client: EntropyDataClient, + resource: Resource, + api_path: str, + entries: list[tuple[str, dict]], + label_prefix: str = "", +) -> SyncResult: """Upsert one resource type via PUT-by-id (idempotent), continue-on-error.""" if resource.topo_sort_parents: - return _upsert_teams(client, resource, entries) + return _upsert_teams(client, resource, api_path, entries) result = SyncResult() for rid, body in entries: payload = {**body, "members": []} if resource.strip_members else body try: - client.put_resource(resource.api_path, rid, payload) - console.print(f" [green]OK[/green] {rid}") + client.put_resource(api_path, rid, payload) + console.print(f" [green]OK[/green] {label_prefix}{rid}") result.ok += 1 if resource.tag_assignments: result.add(_apply_tag_assignments(client, rid, payload)) except ApiError as e: - error_console.print(f" [red]FAIL[/red] {rid}: {e}") - result.fail += 1 - return result - - -def _prune( - client: EntropyDataClient, - resources: list[Resource], - imported_ids: dict[str, set[str]], -) -> SyncResult: - """Delete target resources absent from the import set, in reverse dependency order.""" - result = SyncResult() - for resource in reversed(resources): - keep = imported_ids.get(resource.name, set()) - try: - existing = _list_all(client, resource) - except ApiError as e: - error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + error_console.print(f" [red]FAIL[/red] {label_prefix}{rid}: {e}") result.fail += 1 - continue - - to_delete = [item for item in existing if str(item.get(resource.id_field)) not in keep] - if not to_delete: - continue - - console.print(f"\n[bold]prune {resource.name}[/bold]") - ordered = _order_for_deletion(resource, to_delete) - for item in ordered: - rid = str(item[resource.id_field]) - try: - client.delete_resource(resource.api_path, rid) - console.print(f" [green]DEL[/green] {rid}") - result.ok += 1 - except ApiError as e: - error_console.print(f" [red]FAIL[/red] {rid}: {e}") - result.fail += 1 return result @@ -278,6 +286,69 @@ def _order_for_deletion(resource: Resource, items: list[dict]) -> list[dict]: return ordered +def _prune_one( + client: EntropyDataClient, + resource: Resource, + api_path: str, + keep: set[str], + parent_id: str | None, +) -> SyncResult: + """Delete target items of one (flat resource) or one parent scope (nested) absent from ``keep``.""" + result = SyncResult() + try: + existing = _list_all(client, resource, parent_id) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list {resource.name}: {e}") + result.fail += 1 + return result + + to_delete = [item for item in existing if str(item.get(resource.id_field)) not in keep] + if not to_delete: + return result + + label = f"{resource.name} [{parent_id}]" if parent_id else resource.name + console.print(f"\n[bold]prune {label}[/bold]") + for item in _order_for_deletion(resource, to_delete): + rid = str(item[resource.id_field]) + try: + client.delete_resource(api_path, rid) + console.print(f" [green]DEL[/green] {rid}") + result.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {rid}: {e}") + result.fail += 1 + return result + + +def _prune( + client: EntropyDataClient, + resources: list[Resource], + imported_ids: dict[str, object], +) -> SyncResult: + """Delete target resources absent from the import set, in reverse dependency order. + + For flat resources ``imported_ids[name]`` is a ``set[str]``; for nested resources it + is a ``dict[parent_id, set[child_id]]``. + """ + result = SyncResult() + for resource in reversed(resources): + if resource.parent is None: + keep = imported_ids.get(resource.name) or set() + result.add(_prune_one(client, resource, resource.api_path, keep, None)) + continue + + pairs: dict = imported_ids.get(resource.name) or {} + try: + parent_ids = _parent_ids(client, resource) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list parents for {resource.name}: {e}") + result.fail += 1 + continue + for pid in parent_ids: + result.add(_prune_one(client, resource, resource.path_for(pid), pairs.get(pid, set()), pid)) + return result + + @dataclass class ImportPlan: """Result of a dry-run: per-resource create/update/prune counts.""" @@ -285,6 +356,17 @@ class ImportPlan: counts: dict[str, PlanCounts] = field(default_factory=dict) +def _load_nested(source_dir: Path, resource: Resource) -> dict[str, set[str]]: + """Load a nested resource's artifact into ``{parent_id: {child_id, ...}}``.""" + pairs: dict[str, set[str]] = {} + if not source_dir.is_dir(): + return pairs + for parent_dir in sorted(p for p in source_dir.iterdir() if p.is_dir()): + entries = _load_dir(parent_dir, resource) + pairs[parent_dir.name] = {rid for rid, _ in entries} + return pairs + + def plan_import( client: EntropyDataClient, source: Path, @@ -295,20 +377,37 @@ def plan_import( plan = ImportPlan() for resource in resources: resource_dir = source / resource.name - entries = _load_dir(resource_dir, resource) if resource_dir.is_dir() else [] - imported_ids = {rid for rid, _ in entries} - try: - existing_ids = _enumerate_ids(client, resource) - except ApiError as e: - error_console.print(f"[red]FAIL[/red] list {resource.name}: {e}") - existing_ids = set() - - counts = PlanCounts( - create=len(imported_ids - existing_ids), - update=len(imported_ids & existing_ids), - prune=len(existing_ids - imported_ids) if prune else 0, - ) + if resource.parent is None: + entries = _load_dir(resource_dir, resource) if resource_dir.is_dir() else [] + imported = {rid for rid, _ in entries} + try: + existing = _enumerate_ids(client, resource) + except ApiError as e: + error_console.print(f"[red]FAIL[/red] list {resource.name}: {e}") + existing = set() + counts = PlanCounts( + create=len(imported - existing), + update=len(imported & existing), + prune=len(existing - imported) if prune else 0, + ) + else: + imported_pairs = _load_nested(resource_dir, resource) + existing_pairs: dict[str, set[str]] = {} + try: + for pid in _parent_ids(client, resource): + existing_pairs[pid] = _enumerate_ids(client, resource, pid) + except ApiError as e: + error_console.print(f"[red]FAIL[/red] list {resource.name}: {e}") + counts = PlanCounts() + for pid in set(imported_pairs) | set(existing_pairs): + imp = imported_pairs.get(pid, set()) + exi = existing_pairs.get(pid, set()) + counts.create += len(imp - exi) + counts.update += len(imp & exi) + if prune: + counts.prune += len(exi - imp) + if counts.create or counts.update or counts.prune: plan.counts[resource.name] = counts return plan @@ -322,18 +421,35 @@ def import_dir( ) -> SyncResult: """Upsert every resource under ``source`` in dependency order, optionally pruning.""" total = SyncResult() - imported_ids: dict[str, set[str]] = {} + imported_ids: dict[str, object] = {} for resource in resources: resource_dir = source / resource.name if not resource_dir.is_dir(): continue - entries = _load_dir(resource_dir, resource) - imported_ids[resource.name] = {rid for rid, _ in entries} - if not entries: + + if resource.parent is None: + entries = _load_dir(resource_dir, resource) + imported_ids[resource.name] = {rid for rid, _ in entries} + if not entries: + continue + console.print(f"\n[bold]{resource.name}[/bold]") + total.add(_upsert(client, resource, resource.api_path, entries)) continue - console.print(f"\n[bold]{resource.name}[/bold]") - total.add(_upsert(client, resource, entries)) + + pairs: dict[str, set[str]] = {} + header_printed = False + for parent_dir in sorted(p for p in resource_dir.iterdir() if p.is_dir()): + pid = parent_dir.name + entries = _load_dir(parent_dir, resource) + pairs[pid] = {rid for rid, _ in entries} + if not entries: + continue + if not header_printed: + console.print(f"\n[bold]{resource.name}[/bold]") + header_printed = True + total.add(_upsert(client, resource, resource.path_for(pid), entries, label_prefix=f"{pid}/")) + imported_ids[resource.name] = pairs if prune: total.add(_prune(client, resources, imported_ids)) diff --git a/tests/commands/test_import_export.py b/tests/commands/test_import_export.py index ea2b940..f5e14e6 100644 --- a/tests/commands/test_import_export.py +++ b/tests/commands/test_import_export.py @@ -58,11 +58,22 @@ def test_new_resources_present_in_order(): "dataproducts", "example-data", "access", + "semantic-namespaces", + "semantic-concepts", + "semantic-relationships", ] -def test_semantics_excluded_from_default_order(): - assert "semantics" not in [r.name for r in RESOURCE_ORDER] +def test_semantics_nested_under_namespaces(): + by_name = {r.name: r for r in RESOURCE_ORDER} + # Concepts and relationships are nested under namespaces and come after them. + assert by_name["semantic-concepts"].parent == "semantic-namespaces" + assert by_name["semantic-relationships"].parent == "semantic-namespaces" + names = [r.name for r in RESOURCE_ORDER] + assert names.index("semantic-namespaces") < names.index("semantic-concepts") + assert names.index("semantic-namespaces") < names.index("semantic-relationships") + # The api_path is a parent template expanded per namespace. + assert by_name["semantic-concepts"].path_for("core") == "semantics/experimental/namespaces/core/concepts" def test_select_resources_include_exclude(): @@ -396,3 +407,99 @@ def test_import_help_lists_dir_and_zip(): assert result.exit_code == 0 assert "zip" in result.output assert "dir" in result.output + + +# --- semantics (nested) --------------------------------------------------------- + +NS = f"{BASE_URL}/api/semantics/experimental/namespaces" + + +@responses.activate +def test_export_semantics_nested(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + # One namespace, one concept in it (concept body fetched per-item via detail). + responses.add(responses.GET, NS, json=[{"namespace": "core", "label": "Core"}], status=200) + responses.add(responses.GET, f"{NS}/core/concepts", json=[{"id": "customer"}], status=200) + responses.add( + responses.GET, + f"{NS}/core/concepts/customer", + json={"id": "customer", "name": "Customer", "kind": "entity"}, + status=200, + ) + responses.add(responses.GET, f"{NS}/core/relationships", json=[], status=200) + + dest = tmp_path / "export" + result = runner.invoke( + app, + ["export", "dir", str(dest), "--include", "semantic-namespaces,semantic-concepts,semantic-relationships"], + ) + assert result.exit_code == 0, result.output + + assert yaml.safe_load((dest / "semantic-namespaces" / "core.yaml").read_text())["namespace"] == "core" + # Nested layout: //.yaml, full body from the detail GET. + concept = yaml.safe_load((dest / "semantic-concepts" / "core" / "customer.yaml").read_text()) + assert concept["name"] == "Customer" + + +@responses.activate +def test_import_semantics_nested_namespace_before_concept(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + (src / "semantic-namespaces").mkdir(parents=True) + (src / "semantic-namespaces" / "core.yaml").write_text(yaml.safe_dump({"namespace": "core", "label": "Core"})) + (src / "semantic-concepts" / "core").mkdir(parents=True) + (src / "semantic-concepts" / "core" / "customer.yaml").write_text( + yaml.safe_dump({"id": "customer", "name": "Customer"}) + ) + + puts = [] + + def _record(request): + puts.append(request.url) + return (200, {}, "") + + responses.add_callback(responses.PUT, f"{NS}/core", callback=_record) + responses.add_callback(responses.PUT, f"{NS}/core/concepts/customer", callback=_record) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "semantic-namespaces,semantic-concepts"]) + assert result.exit_code == 0, result.output + # Namespace is created before the concept nested under it. + assert puts == [f"{NS}/core", f"{NS}/core/concepts/customer"] + + +@responses.activate +def test_prune_semantics_nested_deletes_absent_concept(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + (src / "semantic-namespaces").mkdir(parents=True) + (src / "semantic-namespaces" / "core.yaml").write_text(yaml.safe_dump({"namespace": "core"})) + (src / "semantic-concepts" / "core").mkdir(parents=True) + (src / "semantic-concepts" / "core" / "keep.yaml").write_text(yaml.safe_dump({"id": "keep"})) + + responses.add(responses.PUT, f"{NS}/core", status=200) + responses.add(responses.PUT, f"{NS}/core/concepts/keep", status=200) + # Prune enumerates parents (namespaces) then children per namespace on the target. + responses.add(responses.GET, NS, json=[{"namespace": "core"}], status=200) + responses.add(responses.GET, f"{NS}/core/concepts", json=[{"id": "keep"}, {"id": "drop"}], status=200) + + deletes = [] + + def _record_delete(request): + deletes.append(request.url) + return (200, {}, "") + + responses.add_callback(responses.DELETE, f"{NS}/core/concepts/drop", callback=_record_delete) + + result = runner.invoke( + app, + ["import", "dir", str(src), "--include", "semantic-namespaces,semantic-concepts", "--prune", "--yes"], + ) + assert result.exit_code == 0, result.output + # Only the concept absent from the import is deleted, addressed under its namespace. + assert deletes == [f"{NS}/core/concepts/drop"] From e76cd06a7f6fd1c31dd50cdeb70a3fb644f299ed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Deuerling?= Date: Tue, 14 Jul 2026 10:36:45 +0200 Subject: [PATCH 3/4] Copy organization/features as a singleton in export/import/apply Add singleton resource support to the copy engine and include the organization feature configuration (GET/PUT /api/organization/features) as an org-level singleton with no id and no collection. Export reads the object and writes organization-features/ organization-features.yaml (skipping when unset); import PUTs it back; it is applied last so a restrictive policy it carries cannot reject tag/asset imports running earlier, and it is never pruned. Requires the app-side endpoint (entropy-data#1521), which must be merged and deployed to the target instance before this resource can be applied. --- CHANGELOG.md | 1 + README.md | 9 ++- src/entropy_data/resources.py | 26 ++++++-- src/entropy_data/sync.py | 68 ++++++++++++++++++- tests/commands/test_import_export.py | 98 ++++++++++++++++++++++++++++ 5 files changed, 193 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3c99f68..ed29e8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,7 @@ - Add `entropy-data apply --source --to ` to copy portable organization state from one connection to another by exporting the source into a staged directory and importing it into the target. Supports `--prune`, `--dry-run` (per-resource create/update/prune counts), `--include`/`--exclude`, and `--keep ` to retain the staged export. - Fix `import` silently dropping `sourcesystems/` from an organization export — source systems are now imported in their correct dependency position (after policies, before assets). - `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, `example-data`, and the `semantics` graph (namespaces, and concepts/relationships nested per namespace). All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. Nested semantics resources are exported to `//.yaml`, imported namespace-first, and pruned per namespace. +- `export`/`import`/`apply` copy the organization's feature configuration as a singleton (`GET`/`PUT /api/organization/features`, artifact `organization-features/organization-features.yaml`), applied last so a restrictive policy it carries cannot reject earlier imports. Requires the app-side endpoint (entropy-data#1521); singletons are never pruned. ## [0.3.19] diff --git a/README.md b/README.md index 950ff12..89d2fc1 100644 --- a/README.md +++ b/README.md @@ -120,8 +120,9 @@ entropy-data [--version] [--connection NAME] [--output table|json|yaml] [--debug `export`, `import` and `apply` move the portable declarative state of an organization (teams, tags, definitions, policies, source systems, certifications, classification -schemes, assets, data contracts, data products, example data, access agreements, and the -semantics graph) between Entropy Data instances — for example to promote a test environment to prod. +schemes, assets, data contracts, data products, example data, access agreements, the +semantics graph, and the organization feature configuration) between Entropy Data +instances — for example to promote a test environment to prod. Only state reachable through the public `/api/**` API and portable across instances is copied (no secrets, telemetry, or environment-specific identity). Every write is an idempotent PUT-by-id, so runs converge and are safe to repeat. @@ -157,6 +158,10 @@ Notes: keeps them so the artifact is a faithful snapshot. - The semantics graph is nested: namespaces are copied first, then concepts and relationships per namespace. Its artifact lives at `semantic-//.yaml`. +- The organization feature configuration is an org-level singleton + (`organization-features/organization-features.yaml`), applied last and never pruned. It + requires the app's `GET`/`PUT /api/organization/features` endpoint (entropy-data#1521), + which must be merged and deployed to the target instance first. ## Connection Management diff --git a/src/entropy_data/resources.py b/src/entropy_data/resources.py index 1af0918..e16cc04 100644 --- a/src/entropy_data/resources.py +++ b/src/entropy_data/resources.py @@ -10,16 +10,23 @@ teams -> tags -> definitions -> policies -> sourcesystems -> certifications -> classification-schemes -> assets -> datacontracts -> dataproducts -> example-data -> access -> - semantic-namespaces -> semantic-concepts -> semantic-relationships + semantic-namespaces -> semantic-concepts -> semantic-relationships -> + organization-features Pruning walks this list in reverse so dependents are removed before their dependencies. -Most resources are flat: ``/api/{api_path}`` lists the whole organization and -``/api/{api_path}/{id}`` addresses one. A nested resource (``parent`` set) lives -under a parent's path — its ``api_path`` is a ``{parent}`` template expanded per -parent id, and it is enumerated by listing the parent, then listing children for -each parent. Its artifact layout is ``//.yaml``. +Three resource shapes: + +* Flat — ``/api/{api_path}`` lists the whole organization and + ``/api/{api_path}/{id}`` addresses one. Artifact: ``/.yaml``. +* Nested (``parent`` set) — lives under a parent's path; ``api_path`` is a + ``{parent}`` template expanded per parent id, enumerated by listing the parent + then children per parent. Artifact: ``//.yaml``. +* Singleton (``singleton`` set) — one org-level object read/written directly at + ``/api/{api_path}`` (no id, no list). Artifact: ``/.yaml``. Not + prunable. ``organization-features`` is applied last so a restrictive policy it + carries cannot reject the resources imported before it. """ from dataclasses import dataclass @@ -48,6 +55,8 @@ class Resource: endpoint ``PUT /assets/{id}/assigned-tags/{tagId}`` parent name of the parent resource; when set, ``api_path`` is a ``{parent}`` template expanded per parent id (nested resource) + singleton one org-level object at ``/api/{api_path}`` (no id, no list, not + prunable) rather than a collection """ name: str @@ -59,6 +68,7 @@ class Resource: topo_sort_parents: bool = False tag_assignments: bool = False parent: str | None = None + singleton: bool = False def path_for(self, parent_id: str | None = None) -> str: """Resolve ``api_path`` for a given parent id (identity for flat resources).""" @@ -100,6 +110,10 @@ def path_for(self, parent_id: str | None = None) -> str: parent="semantic-namespaces", detail=True, ), + # Org-level singleton. Applied last: it may carry a restrictive managedTagsPolicy + # that would otherwise reject tag/asset imports running earlier in the same apply. + # Requires the app-side GET/PUT /api/organization/features endpoint (entropy-data#1521). + Resource("organization-features", "organization/features", singleton=True), ] RESOURCE_BY_NAME: dict[str, Resource] = {r.name: r for r in RESOURCE_ORDER} diff --git a/src/entropy_data/sync.py b/src/entropy_data/sync.py index d616b35..0deb003 100644 --- a/src/entropy_data/sync.py +++ b/src/entropy_data/sync.py @@ -85,6 +85,27 @@ def _parent_ids(client: EntropyDataClient, resource: Resource) -> list[str]: return [str(item[parent.id_field]) for item in _list_all(client, parent) if item.get(parent.id_field)] +# --- singleton (org-level object, no id) --------------------------------------- + + +def _get_singleton(client: EntropyDataClient, resource: Resource) -> dict: + """GET the singleton object at ``/api/{api_path}`` (returns ``{}`` when unset).""" + response = client.session.get(f"{client.base_url}/api/{resource.api_path}", timeout=REQUEST_TIMEOUT) + _raise_for_status(response) + body = response.json() + return body if isinstance(body, dict) else {} + + +def _put_singleton(client: EntropyDataClient, resource: Resource, body: dict) -> None: + """PUT the singleton object at ``/api/{api_path}`` (no id segment).""" + response = client.session.put(f"{client.base_url}/api/{resource.api_path}", json=body, timeout=REQUEST_TIMEOUT) + _raise_for_status(response) + + +def _singleton_file(root: Path, resource: Resource) -> Path: + return root / resource.name / f"{resource.name}.yaml" + + # --- assigned-tags sub-resource (assets) --------------------------------------- @@ -174,11 +195,36 @@ def _export_items(client: EntropyDataClient, resource: Resource, parent_id: str return total +def _export_singleton(client: EntropyDataClient, resource: Resource, target_dir: Path) -> SyncResult: + """Fetch the singleton object and write it to ``/.yaml`` (skips when unset).""" + total = SyncResult() + try: + body = strip_audit_fields(_get_singleton(client, resource)) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {resource.name}: {e}") + total.fail += 1 + return total + + if not body: # nothing configured on the source; nothing to copy + return total + + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / f"{resource.name}.yaml").write_text( + yaml.safe_dump(body, sort_keys=False, allow_unicode=True, width=4096) + ) + console.print(f" [green]OK[/green] {resource.name}") + total.ok += 1 + return total + + def export_dir(client: EntropyDataClient, dest: Path, resources: list[Resource]) -> SyncResult: """Enumerate the source and write one YAML file per resource under ``dest``.""" total = SyncResult() for resource in resources: console.print(f"\n[bold]{resource.name}[/bold]") + if resource.singleton: + total.add(_export_singleton(client, resource, dest / resource.name)) + continue if resource.parent is None: total.add(_export_items(client, resource, None, dest / resource.name)) continue @@ -332,6 +378,8 @@ def _prune( """ result = SyncResult() for resource in reversed(resources): + if resource.singleton: + continue # a singleton is one object, not a collection — nothing to prune if resource.parent is None: keep = imported_ids.get(resource.name) or set() result.add(_prune_one(client, resource, resource.api_path, keep, None)) @@ -378,7 +426,10 @@ def plan_import( for resource in resources: resource_dir = source / resource.name - if resource.parent is None: + if resource.singleton: + # A singleton is one in-place PUT; count it as an update when the artifact has it. + counts = PlanCounts(update=1) if _singleton_file(source, resource).is_file() else PlanCounts() + elif resource.parent is None: entries = _load_dir(resource_dir, resource) if resource_dir.is_dir() else [] imported = {rid for rid, _ in entries} try: @@ -428,6 +479,21 @@ def import_dir( if not resource_dir.is_dir(): continue + if resource.singleton: + f = _singleton_file(source, resource) + if not f.is_file(): + continue + body = strip_audit_fields(yaml.safe_load(f.read_text()) or {}) + console.print(f"\n[bold]{resource.name}[/bold]") + try: + _put_singleton(client, resource, body) + console.print(f" [green]OK[/green] {resource.name}") + total.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {resource.name}: {e}") + total.fail += 1 + continue + if resource.parent is None: entries = _load_dir(resource_dir, resource) imported_ids[resource.name] = {rid for rid, _ in entries} diff --git a/tests/commands/test_import_export.py b/tests/commands/test_import_export.py index f5e14e6..617b3af 100644 --- a/tests/commands/test_import_export.py +++ b/tests/commands/test_import_export.py @@ -61,9 +61,19 @@ def test_new_resources_present_in_order(): "semantic-namespaces", "semantic-concepts", "semantic-relationships", + "organization-features", ] +def test_organization_features_is_singleton_and_last(): + by_name = {r.name: r for r in RESOURCE_ORDER} + feat = by_name["organization-features"] + assert feat.singleton is True + assert feat.api_path == "organization/features" + # Applied last so a restrictive policy it carries cannot reject earlier imports. + assert RESOURCE_ORDER[-1].name == "organization-features" + + def test_semantics_nested_under_namespaces(): by_name = {r.name: r for r in RESOURCE_ORDER} # Concepts and relationships are nested under namespaces and come after them. @@ -503,3 +513,91 @@ def _record_delete(request): assert result.exit_code == 0, result.output # Only the concept absent from the import is deleted, addressed under its namespace. assert deletes == [f"{NS}/core/concepts/drop"] + + +# --- organization features (singleton) ----------------------------------------- + +FEATURES = f"{BASE_URL}/api/organization/features" + + +@responses.activate +def test_export_organization_features_singleton(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + responses.add( + responses.GET, + FEATURES, + json={"changeProcessMode": "approval-required", "intelligenceEnabled": True, "createdAt": "2020-01-01"}, + status=200, + ) + + dest = tmp_path / "export" + result = runner.invoke(app, ["export", "dir", str(dest), "--include", "organization-features"]) + assert result.exit_code == 0, result.output + + # Singleton layout: /.yaml, audit fields stripped. + body = yaml.safe_load((dest / "organization-features" / "organization-features.yaml").read_text()) + assert body["changeProcessMode"] == "approval-required" + assert "createdAt" not in body + + +@responses.activate +def test_export_organization_features_skips_when_empty(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + # Nothing configured on the source -> empty object -> no file written. + responses.add(responses.GET, FEATURES, json={}, status=200) + + dest = tmp_path / "export" + result = runner.invoke(app, ["export", "dir", str(dest), "--include", "organization-features"]) + assert result.exit_code == 0, result.output + assert not (dest / "organization-features").exists() + + +@responses.activate +def test_import_organization_features_singleton_put(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + (src / "organization-features").mkdir(parents=True) + (src / "organization-features" / "organization-features.yaml").write_text( + yaml.safe_dump({"changeProcessMode": "approval-required", "openInOptions": ["snowflake"]}) + ) + + captured = {} + + def _capture(request): + import json as _json + + captured["body"] = _json.loads(request.body) + return (200, {}, "") + + # PUT to the bare singleton path (no id segment). + responses.add_callback(responses.PUT, FEATURES, callback=_capture) + + result = runner.invoke(app, ["import", "dir", str(src), "--include", "organization-features"]) + assert result.exit_code == 0, result.output + assert captured["body"]["changeProcessMode"] == "approval-required" + assert captured["body"]["openInOptions"] == ["snowflake"] + + +@responses.activate +def test_prune_skips_singleton(monkeypatch, tmp_path): + monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") + monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + + src = tmp_path / "tree" + (src / "organization-features").mkdir(parents=True) + (src / "organization-features" / "organization-features.yaml").write_text( + yaml.safe_dump({"changeProcessMode": "direct-editing-allowed"}) + ) + + responses.add(responses.PUT, FEATURES, status=200) + + # --prune must not attempt to list or delete the singleton (no GET/DELETE registered); + # a stray call would raise ConnectionError and fail the run. + result = runner.invoke(app, ["import", "dir", str(src), "--include", "organization-features", "--prune", "--yes"]) + assert result.exit_code == 0, result.output From 8c483626153ccefa9e0eebd5851ec0e1e4e08a8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Deuerling?= Date: Tue, 14 Jul 2026 14:45:14 +0200 Subject: [PATCH 4/4] Copy semantics as one OSI ontology YAML document per namespace Replace the per-concept/relationship nested handling with a single per-namespace ontology document, using the app's new GET/PUT /api/semantics/experimental/namespaces/{ns}/ontology.yaml endpoint. The app imports the document in the correct internal order (groups before members, concepts before relationships), so the client no longer needs any concept-ordering logic. Add a 'document' resource shape (one raw-YAML document per parent, artifact /.yaml, not prunable) and drop the now-unused nested-collection machinery. The namespace row is still copied first as a normal resource. Requires the app-side ontology.yaml endpoint (entropy-data#1524). --- CHANGELOG.md | 3 +- README.md | 6 +- src/entropy_data/resources.py | 37 +++--- src/entropy_data/sync.py | 169 ++++++++++++++------------- tests/commands/test_import_export.py | 101 +++++----------- 5 files changed, 143 insertions(+), 173 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ed29e8d..e1a5834 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,7 +6,8 @@ - Extend `entropy-data import` with an `import dir ` subcommand alongside `import zip` (the zip form now extracts to a temp directory and reuses the same dir-import logic). Both gain `--prune` to delete target resources absent from the import (in reverse dependency order, guarded by a confirmation prompt unless `--yes`), plus `--include`/`--exclude` filters; `import dir` also supports `--dry-run`. - Add `entropy-data apply --source --to ` to copy portable organization state from one connection to another by exporting the source into a staged directory and importing it into the target. Supports `--prune`, `--dry-run` (per-resource create/update/prune counts), `--include`/`--exclude`, and `--keep ` to retain the staged export. - Fix `import` silently dropping `sourcesystems/` from an organization export — source systems are now imported in their correct dependency position (after policies, before assets). -- `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, `example-data`, and the `semantics` graph (namespaces, and concepts/relationships nested per namespace). All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. Nested semantics resources are exported to `//.yaml`, imported namespace-first, and pruned per namespace. +- `export`/`import`/`apply` now also copy `certifications`, `classification-schemes`, `example-data`, and the `semantics` graph. All writes are idempotent PUT-by-id, team members are stripped on import, asset tag assignments are replayed, and audit fields are stripped before PUT. +- The `semantics` graph is copied as one OSI-compliant ontology YAML document per namespace (`semantic-ontology/.yaml`) via the app's `GET`/`PUT /api/semantics/experimental/namespaces/{ns}/ontology.yaml` endpoint, which imports it in the correct internal order (groups before members, concepts before relationships). Requires that endpoint on the target instance; the namespace row itself is copied first as a normal resource. - `export`/`import`/`apply` copy the organization's feature configuration as a singleton (`GET`/`PUT /api/organization/features`, artifact `organization-features/organization-features.yaml`), applied last so a restrictive policy it carries cannot reject earlier imports. Requires the app-side endpoint (entropy-data#1521); singletons are never pruned. ## [0.3.19] diff --git a/README.md b/README.md index 89d2fc1..d9b2f90 100644 --- a/README.md +++ b/README.md @@ -156,8 +156,10 @@ Notes: - Team members are stripped on import (users are per-instance identities); the export keeps them so the artifact is a faithful snapshot. -- The semantics graph is nested: namespaces are copied first, then concepts and - relationships per namespace. Its artifact lives at `semantic-//.yaml`. +- The semantics graph is copied as one OSI ontology YAML document per namespace + (`semantic-ontology/.yaml`) via the app's `.../{ns}/ontology.yaml` endpoint, + which imports it in the correct internal dependency order. The namespace row is copied + first as a normal resource. Requires that endpoint on the target instance. - The organization feature configuration is an org-level singleton (`organization-features/organization-features.yaml`), applied last and never pruned. It requires the app's `GET`/`PUT /api/organization/features` endpoint (entropy-data#1521), diff --git a/src/entropy_data/resources.py b/src/entropy_data/resources.py index e16cc04..352c751 100644 --- a/src/entropy_data/resources.py +++ b/src/entropy_data/resources.py @@ -10,8 +10,7 @@ teams -> tags -> definitions -> policies -> sourcesystems -> certifications -> classification-schemes -> assets -> datacontracts -> dataproducts -> example-data -> access -> - semantic-namespaces -> semantic-concepts -> semantic-relationships -> - organization-features + semantic-namespaces -> semantic-ontology -> organization-features Pruning walks this list in reverse so dependents are removed before their dependencies. @@ -20,9 +19,12 @@ * Flat — ``/api/{api_path}`` lists the whole organization and ``/api/{api_path}/{id}`` addresses one. Artifact: ``/.yaml``. -* Nested (``parent`` set) — lives under a parent's path; ``api_path`` is a - ``{parent}`` template expanded per parent id, enumerated by listing the parent - then children per parent. Artifact: ``//.yaml``. +* Document (``document`` set, with a ``{parent}`` ``api_path``) — one raw-YAML + document per parent, GET/PUT directly at the expanded path. Enumerated by + listing the parent. Artifact: ``/.yaml``. Not prunable. Used + for the semantic ontology: the app imports the whole namespace document in the + correct internal order (groups before members), so the client needs no + per-concept ordering of its own. * Singleton (``singleton`` set) — one org-level object read/written directly at ``/api/{api_path}`` (no id, no list). Artifact: ``/.yaml``. Not prunable. ``organization-features`` is applied last so a restrictive policy it @@ -54,9 +56,11 @@ class Resource: tag_assignments after PUT, replay ``assignedTags`` via the sub-resource endpoint ``PUT /assets/{id}/assigned-tags/{tagId}`` parent name of the parent resource; when set, ``api_path`` is a - ``{parent}`` template expanded per parent id (nested resource) + ``{parent}`` template expanded per parent id singleton one org-level object at ``/api/{api_path}`` (no id, no list, not prunable) rather than a collection + document one raw-YAML document per parent, GET/PUT directly at the expanded + ``{parent}`` path (not prunable); requires ``parent`` """ name: str @@ -69,13 +73,14 @@ class Resource: tag_assignments: bool = False parent: str | None = None singleton: bool = False + document: bool = False def path_for(self, parent_id: str | None = None) -> str: """Resolve ``api_path`` for a given parent id (identity for flat resources).""" if self.parent is None: return self.api_path if parent_id is None: - raise ValueError(f"Resource '{self.name}' is nested under '{self.parent}' and needs a parent id.") + raise ValueError(f"Resource '{self.name}' is under '{self.parent}' and needs a parent id.") return self.api_path.format(parent=parent_id) @@ -94,21 +99,15 @@ def path_for(self, parent_id: str | None = None) -> str: Resource("dataproducts", "dataproducts", detail=True), Resource("example-data", "example-data"), Resource("access", "access", paginated=True), - # Semantics is nested: concepts and relationships live under a namespace path - # and are enumerated per namespace. They carry their identity in `id` (which may - # contain "/"); the list endpoints may return partial bodies, so fetch each. + # Semantics: the namespace row carries the metadata; the whole ontology (concepts + + # relationships) is copied as one OSI YAML document per namespace via the app's + # ontology.yaml endpoint, which imports it in the correct internal dependency order. Resource("semantic-namespaces", "semantics/experimental/namespaces", id_field="namespace"), Resource( - "semantic-concepts", - "semantics/experimental/namespaces/{parent}/concepts", + "semantic-ontology", + "semantics/experimental/namespaces/{parent}/ontology.yaml", parent="semantic-namespaces", - detail=True, - ), - Resource( - "semantic-relationships", - "semantics/experimental/namespaces/{parent}/relationships", - parent="semantic-namespaces", - detail=True, + document=True, ), # Org-level singleton. Applied last: it may carry a restrictive managedTagsPolicy # that would otherwise reject tag/asset imports running earlier in the same apply. diff --git a/src/entropy_data/sync.py b/src/entropy_data/sync.py index 0deb003..28a88c5 100644 --- a/src/entropy_data/sync.py +++ b/src/entropy_data/sync.py @@ -5,10 +5,9 @@ and prune implementation and cannot drift. Everything is continue-on-error: a single resource failure is logged and counted, never aborts the run. -Flat resources are `/api/{api_path}` (list-all) + `/api/{api_path}/{id}`. A nested -resource (``Resource.parent`` set) is enumerated by first listing its parent, then -listing children under each parent's expanded path; its artifact lives at -``//.yaml``. +Flat resources are `/api/{api_path}` (list-all) + `/api/{api_path}/{id}`. A document +resource (``Resource.document`` set) is one raw-YAML document per parent, GET/PUT at +the expanded ``{parent}`` path; its artifact lives at ``/.yaml``. """ from dataclasses import dataclass, field @@ -106,6 +105,51 @@ def _singleton_file(root: Path, resource: Resource) -> Path: return root / resource.name / f"{resource.name}.yaml" +# --- document (one raw-YAML document per parent) -------------------------------- + + +def _get_document(client: EntropyDataClient, resource: Resource, parent_id: str) -> str: + """GET the raw YAML document at the expanded ``{parent}`` path.""" + response = client.session.get( + f"{client.base_url}/api/{resource.path_for(parent_id)}", + headers={"Accept": "application/yaml"}, + timeout=REQUEST_TIMEOUT, + ) + _raise_for_status(response) + return response.text + + +def _put_document(client: EntropyDataClient, resource: Resource, parent_id: str, text: str) -> None: + """PUT the raw YAML document to the expanded ``{parent}`` path.""" + response = client.session.put( + f"{client.base_url}/api/{resource.path_for(parent_id)}", + data=text.encode("utf-8"), + headers={"Content-Type": "application/yaml"}, + timeout=REQUEST_TIMEOUT, + ) + _raise_for_status(response) + + +def _export_document(client: EntropyDataClient, resource: Resource, parent_id: str, target_dir: Path) -> SyncResult: + """Fetch one parent's YAML document and write it to ``/.yaml`` (skips when empty).""" + total = SyncResult() + try: + text = _get_document(client, resource, parent_id) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {parent_id}: {e}") + total.fail += 1 + return total + + if not text.strip(): # empty document; nothing to copy + return total + + target_dir.mkdir(parents=True, exist_ok=True) + (target_dir / f"{sanitize_filename(parent_id)}.yaml").write_text(text) + console.print(f" [green]OK[/green] {parent_id}") + total.ok += 1 + return total + + # --- assigned-tags sub-resource (assets) --------------------------------------- @@ -225,18 +269,17 @@ def export_dir(client: EntropyDataClient, dest: Path, resources: list[Resource]) if resource.singleton: total.add(_export_singleton(client, resource, dest / resource.name)) continue - if resource.parent is None: - total.add(_export_items(client, resource, None, dest / resource.name)) - continue - - try: - parent_ids = _parent_ids(client, resource) - except ApiError as e: - error_console.print(f" [red]FAIL[/red] list parents for {resource.name}: {e}") - total.fail += 1 + if resource.document: + try: + parent_ids = _parent_ids(client, resource) + except ApiError as e: + error_console.print(f" [red]FAIL[/red] list parents for {resource.name}: {e}") + total.fail += 1 + continue + for pid in parent_ids: + total.add(_export_document(client, resource, pid, dest / resource.name)) continue - for pid in parent_ids: - total.add(_export_items(client, resource, pid, dest / resource.name / sanitize_filename(pid))) + total.add(_export_items(client, resource, None, dest / resource.name)) return total @@ -373,27 +416,14 @@ def _prune( ) -> SyncResult: """Delete target resources absent from the import set, in reverse dependency order. - For flat resources ``imported_ids[name]`` is a ``set[str]``; for nested resources it - is a ``dict[parent_id, set[child_id]]``. + ``imported_ids[name]`` is a ``set[str]`` of the flat resource's imported ids. """ result = SyncResult() for resource in reversed(resources): - if resource.singleton: - continue # a singleton is one object, not a collection — nothing to prune - if resource.parent is None: - keep = imported_ids.get(resource.name) or set() - result.add(_prune_one(client, resource, resource.api_path, keep, None)) - continue - - pairs: dict = imported_ids.get(resource.name) or {} - try: - parent_ids = _parent_ids(client, resource) - except ApiError as e: - error_console.print(f" [red]FAIL[/red] list parents for {resource.name}: {e}") - result.fail += 1 - continue - for pid in parent_ids: - result.add(_prune_one(client, resource, resource.path_for(pid), pairs.get(pid, set()), pid)) + if resource.singleton or resource.document: + continue # one object / one document — not a prunable collection + keep = imported_ids.get(resource.name) or set() + result.add(_prune_one(client, resource, resource.api_path, keep, None)) return result @@ -404,17 +434,6 @@ class ImportPlan: counts: dict[str, PlanCounts] = field(default_factory=dict) -def _load_nested(source_dir: Path, resource: Resource) -> dict[str, set[str]]: - """Load a nested resource's artifact into ``{parent_id: {child_id, ...}}``.""" - pairs: dict[str, set[str]] = {} - if not source_dir.is_dir(): - return pairs - for parent_dir in sorted(p for p in source_dir.iterdir() if p.is_dir()): - entries = _load_dir(parent_dir, resource) - pairs[parent_dir.name] = {rid for rid, _ in entries} - return pairs - - def plan_import( client: EntropyDataClient, source: Path, @@ -429,7 +448,11 @@ def plan_import( if resource.singleton: # A singleton is one in-place PUT; count it as an update when the artifact has it. counts = PlanCounts(update=1) if _singleton_file(source, resource).is_file() else PlanCounts() - elif resource.parent is None: + elif resource.document: + # One in-place PUT per parent document present in the artifact. + n = len(list(resource_dir.glob("*.yaml"))) if resource_dir.is_dir() else 0 + counts = PlanCounts(update=n) + else: entries = _load_dir(resource_dir, resource) if resource_dir.is_dir() else [] imported = {rid for rid, _ in entries} try: @@ -442,22 +465,6 @@ def plan_import( update=len(imported & existing), prune=len(existing - imported) if prune else 0, ) - else: - imported_pairs = _load_nested(resource_dir, resource) - existing_pairs: dict[str, set[str]] = {} - try: - for pid in _parent_ids(client, resource): - existing_pairs[pid] = _enumerate_ids(client, resource, pid) - except ApiError as e: - error_console.print(f"[red]FAIL[/red] list {resource.name}: {e}") - counts = PlanCounts() - for pid in set(imported_pairs) | set(existing_pairs): - imp = imported_pairs.get(pid, set()) - exi = existing_pairs.get(pid, set()) - counts.create += len(imp - exi) - counts.update += len(imp & exi) - if prune: - counts.prune += len(exi - imp) if counts.create or counts.update or counts.prune: plan.counts[resource.name] = counts @@ -494,28 +501,28 @@ def import_dir( total.fail += 1 continue - if resource.parent is None: - entries = _load_dir(resource_dir, resource) - imported_ids[resource.name] = {rid for rid, _ in entries} - if not entries: - continue - console.print(f"\n[bold]{resource.name}[/bold]") - total.add(_upsert(client, resource, resource.api_path, entries)) + if resource.document: + header_printed = False + for f in sorted(resource_dir.glob("*.yaml")): + parent_id = f.stem + if not header_printed: + console.print(f"\n[bold]{resource.name}[/bold]") + header_printed = True + try: + _put_document(client, resource, parent_id, f.read_text()) + console.print(f" [green]OK[/green] {parent_id}") + total.ok += 1 + except ApiError as e: + error_console.print(f" [red]FAIL[/red] {parent_id}: {e}") + total.fail += 1 continue - pairs: dict[str, set[str]] = {} - header_printed = False - for parent_dir in sorted(p for p in resource_dir.iterdir() if p.is_dir()): - pid = parent_dir.name - entries = _load_dir(parent_dir, resource) - pairs[pid] = {rid for rid, _ in entries} - if not entries: - continue - if not header_printed: - console.print(f"\n[bold]{resource.name}[/bold]") - header_printed = True - total.add(_upsert(client, resource, resource.path_for(pid), entries, label_prefix=f"{pid}/")) - imported_ids[resource.name] = pairs + entries = _load_dir(resource_dir, resource) + imported_ids[resource.name] = {rid for rid, _ in entries} + if not entries: + continue + console.print(f"\n[bold]{resource.name}[/bold]") + total.add(_upsert(client, resource, resource.api_path, entries)) if prune: total.add(_prune(client, resources, imported_ids)) diff --git a/tests/commands/test_import_export.py b/tests/commands/test_import_export.py index 617b3af..02efa45 100644 --- a/tests/commands/test_import_export.py +++ b/tests/commands/test_import_export.py @@ -59,8 +59,7 @@ def test_new_resources_present_in_order(): "example-data", "access", "semantic-namespaces", - "semantic-concepts", - "semantic-relationships", + "semantic-ontology", "organization-features", ] @@ -74,16 +73,15 @@ def test_organization_features_is_singleton_and_last(): assert RESOURCE_ORDER[-1].name == "organization-features" -def test_semantics_nested_under_namespaces(): +def test_semantic_ontology_is_document_under_namespaces(): by_name = {r.name: r for r in RESOURCE_ORDER} - # Concepts and relationships are nested under namespaces and come after them. - assert by_name["semantic-concepts"].parent == "semantic-namespaces" - assert by_name["semantic-relationships"].parent == "semantic-namespaces" + onto = by_name["semantic-ontology"] + # The whole ontology is one YAML document per namespace, imported after the namespace row. + assert onto.document is True + assert onto.parent == "semantic-namespaces" names = [r.name for r in RESOURCE_ORDER] - assert names.index("semantic-namespaces") < names.index("semantic-concepts") - assert names.index("semantic-namespaces") < names.index("semantic-relationships") - # The api_path is a parent template expanded per namespace. - assert by_name["semantic-concepts"].path_for("core") == "semantics/experimental/namespaces/core/concepts" + assert names.index("semantic-namespaces") < names.index("semantic-ontology") + assert onto.path_for("core") == "semantics/experimental/namespaces/core/ontology.yaml" def test_select_resources_include_exclude(): @@ -419,100 +417,63 @@ def test_import_help_lists_dir_and_zip(): assert "dir" in result.output -# --- semantics (nested) --------------------------------------------------------- +# --- semantics (ontology document per namespace) -------------------------------- NS = f"{BASE_URL}/api/semantics/experimental/namespaces" @responses.activate -def test_export_semantics_nested(monkeypatch, tmp_path): +def test_export_semantics_ontology_document(monkeypatch, tmp_path): monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") - # One namespace, one concept in it (concept body fetched per-item via detail). + ontology = "version: 0.2.0.dev0\nname: core\nontology:\n - concept:\n id: customer\n" responses.add(responses.GET, NS, json=[{"namespace": "core", "label": "Core"}], status=200) - responses.add(responses.GET, f"{NS}/core/concepts", json=[{"id": "customer"}], status=200) responses.add( responses.GET, - f"{NS}/core/concepts/customer", - json={"id": "customer", "name": "Customer", "kind": "entity"}, + f"{NS}/core/ontology.yaml", + body=ontology, status=200, + content_type="application/yaml", ) - responses.add(responses.GET, f"{NS}/core/relationships", json=[], status=200) dest = tmp_path / "export" - result = runner.invoke( - app, - ["export", "dir", str(dest), "--include", "semantic-namespaces,semantic-concepts,semantic-relationships"], - ) + result = runner.invoke(app, ["export", "dir", str(dest), "--include", "semantic-namespaces,semantic-ontology"]) assert result.exit_code == 0, result.output assert yaml.safe_load((dest / "semantic-namespaces" / "core.yaml").read_text())["namespace"] == "core" - # Nested layout: //.yaml, full body from the detail GET. - concept = yaml.safe_load((dest / "semantic-concepts" / "core" / "customer.yaml").read_text()) - assert concept["name"] == "Customer" + # The whole ontology is one document per namespace, written verbatim (no re-serialization). + assert (dest / "semantic-ontology" / "core.yaml").read_text() == ontology @responses.activate -def test_import_semantics_nested_namespace_before_concept(monkeypatch, tmp_path): +def test_import_semantics_ontology_document(monkeypatch, tmp_path): monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") + ontology = "version: 0.2.0.dev0\nname: core\nontology: []\n" src = tmp_path / "tree" (src / "semantic-namespaces").mkdir(parents=True) (src / "semantic-namespaces" / "core.yaml").write_text(yaml.safe_dump({"namespace": "core", "label": "Core"})) - (src / "semantic-concepts" / "core").mkdir(parents=True) - (src / "semantic-concepts" / "core" / "customer.yaml").write_text( - yaml.safe_dump({"id": "customer", "name": "Customer"}) - ) + (src / "semantic-ontology").mkdir(parents=True) + (src / "semantic-ontology" / "core.yaml").write_text(ontology) - puts = [] + captured = {} - def _record(request): - puts.append(request.url) + def _record_ontology(request): + body = request.body + captured["body"] = body.decode("utf-8") if isinstance(body, bytes) else body + captured["content_type"] = request.headers.get("Content-Type") return (200, {}, "") - responses.add_callback(responses.PUT, f"{NS}/core", callback=_record) - responses.add_callback(responses.PUT, f"{NS}/core/concepts/customer", callback=_record) - - result = runner.invoke(app, ["import", "dir", str(src), "--include", "semantic-namespaces,semantic-concepts"]) - assert result.exit_code == 0, result.output - # Namespace is created before the concept nested under it. - assert puts == [f"{NS}/core", f"{NS}/core/concepts/customer"] - - -@responses.activate -def test_prune_semantics_nested_deletes_absent_concept(monkeypatch, tmp_path): - monkeypatch.setattr(cfg, "CONFIG_FILE", tmp_path / "config.toml") - monkeypatch.setenv("ENTROPY_DATA_API_KEY", "test-key") - - src = tmp_path / "tree" - (src / "semantic-namespaces").mkdir(parents=True) - (src / "semantic-namespaces" / "core.yaml").write_text(yaml.safe_dump({"namespace": "core"})) - (src / "semantic-concepts" / "core").mkdir(parents=True) - (src / "semantic-concepts" / "core" / "keep.yaml").write_text(yaml.safe_dump({"id": "keep"})) - responses.add(responses.PUT, f"{NS}/core", status=200) - responses.add(responses.PUT, f"{NS}/core/concepts/keep", status=200) - # Prune enumerates parents (namespaces) then children per namespace on the target. - responses.add(responses.GET, NS, json=[{"namespace": "core"}], status=200) - responses.add(responses.GET, f"{NS}/core/concepts", json=[{"id": "keep"}, {"id": "drop"}], status=200) - - deletes = [] + responses.add_callback(responses.PUT, f"{NS}/core/ontology.yaml", callback=_record_ontology) - def _record_delete(request): - deletes.append(request.url) - return (200, {}, "") - - responses.add_callback(responses.DELETE, f"{NS}/core/concepts/drop", callback=_record_delete) - - result = runner.invoke( - app, - ["import", "dir", str(src), "--include", "semantic-namespaces,semantic-concepts", "--prune", "--yes"], - ) + result = runner.invoke(app, ["import", "dir", str(src), "--include", "semantic-namespaces,semantic-ontology"]) assert result.exit_code == 0, result.output - # Only the concept absent from the import is deleted, addressed under its namespace. - assert deletes == [f"{NS}/core/concepts/drop"] + # The ontology document is PUT verbatim as application/yaml — the app orders it internally. + assert captured["body"] == ontology + assert captured["content_type"] == "application/yaml" # --- organization features (singleton) -----------------------------------------