Add script that allows you to "import" a workspace from another server (HEXA-1663)

.gitignore

@@ -1,3 +1,4 @@
+__pycache__/
 build/
 .env
 .envlocal

script/migrate_lib/connections.py

@@ -0,0 +1,141 @@
+"""Connection migration: copy every workspace connection to the target.
+
+Secret field values come through the source API only because we authenticate
+as a Django superuser: the ``value`` resolver redacts secret fields unless the
+caller has ``workspaces.update_connection``, which a superuser short-circuits
+to True. If a secret still comes back empty, we create the connection anyway
+and warn — the user must set that secret manually on the target.
+
+Connection slugs are preserved: unlike createWorkspace / createPipeline (which
+re-derive the slug/code server-side), createConnection honors a caller-supplied
+slug and only suffixes it on collision. The target workspace is fresh, so the
+source slug carries over intact, keeping pipeline parameters that reference a
+connection by slug valid.
+"""
+
+import sys
+from dataclasses import dataclass, field
+from typing import Any
+
+from openhexa.graphql.graphql_client.client import Client
+from openhexa.graphql.graphql_client.input_types import (
+    ConnectionFieldInput,
+    ConnectionType,
+    CreateConnectionInput,
+)
+
+from .transport import GraphQLError, gql
+
+
+# `workspace.connections` is a (non-paginated) field on Workspace returning the
+# full list; the SDK's workspace() doesn't pull fields, so we query it raw.
+LIST_CONNECTIONS_QUERY = """
+query ListConnections($slug: String!) {
+    workspace(slug: $slug) {
+        connections {
+            id name slug description type
+            fields { code value secret }
+        }
+    }
+}
+"""
+
+
+@dataclass
+class ConnectionsResult:
+    """What migrate_all() did, for the orchestrator to print."""
+
+    created: list[tuple[str, int]] = field(default_factory=list)
+    """(slug, field_count) for each connection created on target."""
+
+    skipped: list[str] = field(default_factory=list)
+    """Slugs that already existed on target."""
+
+    failed: list[str] = field(default_factory=list)
+    """Slugs whose creation failed; user must handle manually."""
+
+    warnings: list[str] = field(default_factory=list)
+    """Human-readable warnings (e.g. secret fields created with no value)."""
+
+
+def _list_connections(client: Client, slug: str) -> list[dict[str, Any]] | None:
+    """Return the workspace's connections, or None if the workspace is absent."""
+    data = gql(client, LIST_CONNECTIONS_QUERY, {"slug": slug}, "ListConnections")
+    ws = data["workspace"]
+    if ws is None:
+        return None
+    return list(ws["connections"])
+
+
+def _build_fields(
+    conn: dict[str, Any], result: ConnectionsResult
+) -> list[ConnectionFieldInput]:
+    """Map source fields to ConnectionFieldInput, warning on empty secrets."""
+    fields_in: list[ConnectionFieldInput] = []
+    for f in conn.get("fields") or []:
+        value = f.get("value")
+        if f.get("secret") and not value:
+            result.warnings.append(
+                f"connection '{conn['slug']}' field '{f['code']}' is a secret "
+                "with no readable value on source — created empty; set it "
+                "manually on the target."
+            )
+        fields_in.append(
+            ConnectionFieldInput(
+                code=f["code"], secret=bool(f.get("secret")), value=value
+            )
+        )
+    return fields_in
+
+
+def migrate_all(
+    source: Client,
+    target: Client,
+    source_slug: str,
+    target_slug: str,
+) -> ConnectionsResult:
+    """Copy every connection from `source_slug` into `target_slug`."""
+    result = ConnectionsResult()
+
+    print("=> Listing source connections ...")
+    conns = _list_connections(source, source_slug)
+    if conns is None:
+        raise GraphQLError(
+            f"source workspace '{source_slug}' not found while listing connections"
+        )
+    print(f"   found {len(conns)} connection(s)")
+
+    existing = {c["slug"] for c in (_list_connections(target, target_slug) or [])}
+
+    for conn in conns:
+        slug = conn["slug"]
+        if slug in existing:
+            print(f"   - connection '{slug}' already exists on target — skipping")
+            result.skipped.append(slug)
+            continue
+        try:
+            print(f"   - migrating connection '{slug}' ...")
+            fields_in = _build_fields(conn, result)
+            res = target.create_connection(
+                input=CreateConnectionInput(
+                    workspace_slug=target_slug,
+                    name=conn["name"],
+                    slug=slug,
+                    type=ConnectionType(conn["type"]),
+                    description=conn.get("description") or "",
+                    fields=fields_in,
+                )
+            )
+            if not res.success or res.connection is None:
+                raise GraphQLError(
+                    f"createConnection failed for '{slug}': "
+                    + ",".join(e.value for e in (res.errors or []))
+                )
+            result.created.append((slug, len(fields_in)))
+        except GraphQLError as exc:
+            # Collect and continue (like files.py) so one bad connection
+            # doesn't abort the rest of the migration.
+            print(f"\tFAILED to migrate connection '{slug}': {exc}", file=sys.stderr)
+            result.failed.append(slug)
+
+    return result

script/migrate_lib/files.py

@@ -0,0 +1,189 @@
+"""Workspace file transfer: list / download / upload, plus full bucket migration."""
+
+import sys
+from collections.abc import Iterator
+from dataclasses import dataclass, field
+from typing import Any
+
+import httpx
+from openhexa.graphql.graphql_client.client import Client
+
+from .transport import GraphQLError, _dbg, gql
+
+
+OBJECTS_PAGE_SIZE = 100
+
+
+PREPARE_DOWNLOAD_MUTATION = """
+mutation PrepareDownload($input: PrepareObjectDownloadInput!) {
+    prepareObjectDownload(input: $input) {
+        success errors downloadUrl
+    }
+}
+"""
+
+PREPARE_UPLOAD_MUTATION = """
+mutation PrepareUpload($input: PrepareObjectUploadInput!) {
+    prepareObjectUpload(input: $input) {
+        success errors uploadUrl headers
+    }
+}
+"""
+
+LIST_OBJECTS_QUERY = """
+query ListObjects($slug: String!, $prefix: String, $page: Int!, $perPage: Int!) {
+    workspace(slug: $slug) {
+        bucket {
+            objects(prefix: $prefix, page: $page, perPage: $perPage, ignoreHiddenFiles: false) {
+                hasNextPage
+                items {
+                    key name path size type
+                }
+            }
+        }
+    }
+}
+"""
+
+
+@dataclass
+class FilesResult:
+    """What migrate_all() did, for the orchestrator to print."""
+
+    copied: list[tuple[str, int]] = field(default_factory=list)
+    """(object_key, byte_size) for each file copied to target."""
+
+    failed: list[str] = field(default_factory=list)
+    """Object keys whose download or upload failed; user must handle manually."""
+
+
+def download(source: Client, ws_slug: str, file_path: str) -> bytes:
+    """Download a file from the source workspace via a presigned URL."""
+    data = gql(
+        source,
+        PREPARE_DOWNLOAD_MUTATION,
+        {
+            "input": {
+                "workspaceSlug": ws_slug,
+                "objectKey": file_path,
+                "forceAttachment": False,
+            }
+        },
+        "PrepareDownload",
+    )
+    result = data["prepareObjectDownload"]
+    if not result["success"] or not result.get("downloadUrl"):
+        raise GraphQLError(
+            f"prepareObjectDownload failed for '{file_path}': "
+            + ",".join(result.get("errors") or [])
+        )
+    url = result["downloadUrl"]
+    _dbg(f"download {file_path} <- {url}")
+    with httpx.Client(timeout=300) as c:
+        resp = c.get(url)
+    if not resp.is_success:
+        raise GraphQLError(
+            f"download of '{file_path}' returned HTTP {resp.status_code}: "
+            f"{resp.text[:500]}"
+        )
+    return resp.content
+
+
+def upload(
+    target: Client,
+    ws_slug: str,
+    file_path: str,
+    content: bytes,
+    content_type: str = "application/octet-stream",
+) -> None:
+    """Upload bytes to the target workspace at the given object key."""
+    data = gql(
+        target,
+        PREPARE_UPLOAD_MUTATION,
+        {
+            "input": {
+                "workspaceSlug": ws_slug,
+                "objectKey": file_path,
+                "contentType": content_type,
+            }
+        },
+        "PrepareUpload",
+    )
+    result = data["prepareObjectUpload"]
+    if not result["success"] or not result.get("uploadUrl"):
+        raise GraphQLError(
+            f"prepareObjectUpload failed for '{file_path}': "
+            + ",".join(result.get("errors") or [])
+        )
+    url = result["uploadUrl"]
+    headers = dict(result.get("headers") or {})
+    headers.setdefault("Content-Type", content_type)
+    _dbg(f"upload {file_path} ({len(content)} bytes) -> {url}")
+    with httpx.Client(timeout=300) as c:
+        resp = c.put(url, content=content, headers=headers)
+    if not resp.is_success:
+        raise GraphQLError(
+            f"upload of '{file_path}' returned HTTP {resp.status_code}: "
+            f"{resp.text[:500]}"
+        )
+
+
+def walk(client: Client, ws_slug: str, prefix: str = "") -> Iterator[dict[str, Any]]:
+    """Recursively yield FILE BucketObject dicts under `prefix`.
+
+    The bucket.objects field is delimited (returns DIRECTORY entries rather
+    than recursing), so we walk each directory ourselves.
+    """
+    # Buffer all directory entries at this level so we don't interleave
+    # recursive listings with this level's pagination.
+    subdirs: list[str] = []
+    page = 1
+    while True:
+        data = gql(
+            client,
+            LIST_OBJECTS_QUERY,
+            {
+                "slug": ws_slug,
+                "prefix": prefix or None,
+                "page": page,
+                "perPage": OBJECTS_PAGE_SIZE,
+            },
+            "ListObjects",
+        )
+        ws = data["workspace"]
+        if ws is None:
+            return
+        page_data = ws["bucket"]["objects"]
+        for obj in page_data["items"]:
+            if obj["type"] == "FILE":
+                yield obj
+            elif obj["type"] == "DIRECTORY":
+                subdirs.append(obj["key"])
+        if not page_data["hasNextPage"]:
+            break
+        page += 1
+    for sub in subdirs:
+        yield from walk(client, ws_slug, sub)
+
+
+def migrate_all(
+    source: Client, target: Client, source_slug: str, target_slug: str
+) -> FilesResult:
+    """Copy every file from the source workspace bucket to the target bucket."""
+    result = FilesResult()
+    print("=> Listing source files ...")
+    for obj in walk(source, source_slug):
+        path = obj["key"]
+        size = obj.get("size") or 0
+        try:
+            print(f"   - copying '{path}' ({size} bytes) ...")
+            content = download(source, source_slug, path)
+            upload(target, target_slug, path, content)
+            result.copied.append((path, len(content)))
+        except GraphQLError:
+            # Surface the failure in the live log without dumping the (often
+            # HTML) server response. The full path goes into result.failed for
+            # the final summary so the user can re-attempt manually.
+            print(f"\tFAILED to copy '{path}'", file=sys.stderr)
+            result.failed.append(path)
+    return result