Create the BigLake Iceberg REST catalog for this bucket if it's missing. The Iceberg REST `/v1/config?warehouse=gs:// ` lookup resolves the catalog whose id equals the bucket name; it does not provision one on demand. Without it, every REST call returns a sanitized 403. Creating
(token: str, project: str, bucket: str)
| 84 | |
| 85 | |
| 86 | def ensure_catalog(token: str, project: str, bucket: str) -> None: |
| 87 | """Create the BigLake Iceberg REST catalog for this bucket if it's missing. |
| 88 | |
| 89 | The Iceberg REST `/v1/config?warehouse=gs://<bucket>` lookup resolves the catalog |
| 90 | whose id equals the bucket name; it does not provision one on demand. Without it, |
| 91 | every REST call returns a sanitized 403. Creating it here lets callers bootstrap |
| 92 | their own catalog instead of depending on out-of-band (Pulumi/console) setup. |
| 93 | |
| 94 | Credential mode is END_USER: the Materialize sink writes to GCS with the service |
| 95 | account's own credentials (see `connect_rest` for the GCP branch in |
| 96 | mz_storage_types::connections), not credentials vended by the catalog. |
| 97 | """ |
| 98 | base = f"{BIGLAKE_REST_BASE}/extensions/projects/{project}/catalogs" |
| 99 | |
| 100 | # GET returns the catalog if it exists, 404 if not. |
| 101 | try: |
| 102 | urllib.request.urlopen( |
| 103 | biglake_request("GET", f"{base}/{bucket}", token, project) |
| 104 | ) |
| 105 | return |
| 106 | except urllib.error.HTTPError as e: |
| 107 | if e.code != 404: |
| 108 | raise |
| 109 | |
| 110 | create_url = f"{base}?iceberg-catalog-id={urllib.parse.quote(bucket, safe='')}" |
| 111 | req = biglake_request("POST", create_url, token, project) |
| 112 | req.add_header("Content-Type", "application/json") |
| 113 | req.data = json.dumps( |
| 114 | { |
| 115 | "catalog-type": "CATALOG_TYPE_GCS_BUCKET", |
| 116 | "credential-mode": "CREDENTIAL_MODE_END_USER", |
| 117 | } |
| 118 | ).encode() |
| 119 | try: |
| 120 | urllib.request.urlopen(req) |
| 121 | print(f"created BigLake catalog for gs://{bucket}") |
| 122 | except urllib.error.HTTPError as e: |
| 123 | # A concurrent run can create the catalog between our GET and POST. |
| 124 | if e.code == 409: |
| 125 | return |
| 126 | body = e.read().decode("utf-8", errors="replace") |
| 127 | print(f"BigLake catalog create failed: HTTP {e.code}\n{body}") |
| 128 | raise |
| 129 | |
| 130 | |
| 131 | def resolve_warehouse_prefix(token: str, project: str, bucket: str) -> str: |
no test coverage detected