Tableflow update patch

src/confluent_sql/__init__.py

@@ -58,6 +58,7 @@
     TableflowPhase,
     TableflowTopic,
     TableflowTopicConfig,
+    TableflowTopicSpec,
     TableFormat,
 )
 from .types import PropertiesDict, PropertiesMapping, SqlNone, YearMonthInterval
@@ -103,6 +104,7 @@
     "TableflowPhase",
     "TableflowTopic",
     "TableflowTopicConfig",
+    "TableflowTopicSpec",
     "ManagedStorage",
     "ByobAwsStorage",
     "AzureAdlsStorage",

src/confluent_sql/connection.py

@@ -53,6 +53,7 @@
     TableflowTopicConfig,
     TableFormat,
     build_create_payload,
+    build_update_payload,
     normalize_table_formats,
 )
 from .types import PropertiesDict, RowPythonTypes, StrAnyDict
@@ -2205,6 +2206,97 @@ def get_tableflow(self, table_name: str) -> TableflowTopic:
             ) from e
         return TableflowTopic.from_response(response.json())
 
+    def update_tableflow(
+        self,
+        table_name: str,
+        *,
+        table_formats: TableFormat | Collection[TableFormat] | None = None,
+        config: TableflowTopicConfig | None = None,
+        wait_for_running: bool = True,
+        timeout: float = 300,
+    ) -> TableflowTopic:
+        """Update an already-enabled Tableflow topic's `table_formats`/`config`.
+
+        `PATCH /tableflow/v1/tableflow-topics/{display_name}`. `table_formats`/`config` are the
+        only fields updatable this way -- `storage`/`display_name` are `x-immutable`; a caller
+        needing to change either has no in-place path and must recreate the topic instead.
+
+        `None` means "leave unchanged," for this method's own two arguments and for each of
+        `config`'s own sub-fields (same convention `config.to_spec()` already uses for create) --
+        not "clear it": none of `retention_ms`/`data_retention_ms`/`error_handling` can actually
+        be unset via this API (each has a server-enforced default, and the server rejects an
+        explicit null for any of them), so there's no supported way to request a delete, and this
+        method doesn't attempt one. This method itself doesn't diff against the topic's current
+        live state -- that's the caller's job (comparing against `get_tableflow`'s response);
+        it just sends whatever `table_formats`/`config` it's given, same as `enable_tableflow`.
+
+        Args:
+            table_name: The Flink table / Kafka topic name (the {display_name} path segment).
+            table_formats: New table_formats, replacing the full list -- or None to leave
+                unchanged.
+            config: New topic-level config -- or None to leave unchanged. Only fields actually
+                set on it (not None) are changed; the rest are left as-is.
+            wait_for_running: If True (default), poll until the topic reaches RUNNING, raising on
+                FAILED. If False, return as soon as the update is accepted.
+            timeout: Maximum seconds to wait when wait_for_running is True.
+
+        Returns:
+            The TableflowTopic -- RUNNING when waited (the default), otherwise the just-accepted
+            topic (typically PENDING while formats/config changes roll out).
+
+        Raises:
+            InterfaceError: If table_formats and config are both None (nothing to update), or if
+                table_formats is empty or names an unknown format.
+            ProgrammingError: If no control-plane credential is available, or the cluster id
+                can't be resolved without a global key.
+            TableflowTopicNotFoundError: If Tableflow is not enabled for the topic (HTTP 404).
+            OperationalError: On other API errors, on FAILED during a wait, or on wait timeout.
+        """
+        # Validate/normalize before any network work (including the possibly CMK-resolving
+        # cluster-id lookup below), same as enable_tableflow.
+        wire_formats = normalize_table_formats(table_formats) if table_formats is not None else None
+        config_spec = config.to_spec() if config is not None else None
+        if not wire_formats and not config_spec:
+            raise InterfaceError("update_tableflow requires at least one of table_formats/config")
+
+        payload = build_update_payload(
+            table_formats=wire_formats,
+            config_spec=config_spec,
+            environment_id=self.environment_id,
+            kafka_cluster_id=self._resolve_kafka_cluster_id(),
+        )
+        logger.info(f"Updating Tableflow for table '{table_name}'")
+        response = self._tableflow_request(
+            f"{self._TABLEFLOW_TOPICS_PATH}/{table_name}",
+            method="PATCH",
+            json=payload,
+            raise_for_status=False,
+        )
+        try:
+            response.raise_for_status()
+        except httpx.HTTPStatusError as e:
+            if e.response.status_code == 404:
+                raise TableflowTopicNotFoundError(
+                    f"Tableflow is not enabled for table '{table_name}'", table_name=table_name
+                ) from e
+            try:
+                res = e.response.json()
+                errors = res.get("errors", [])
+                details = "; ".join(err["detail"] for err in errors if err.get("detail"))
+            except Exception:
+                details = ""
+            if not details:
+                details = "no more details"
+            raise OperationalError(
+                f"Error updating Tableflow topic: {details}",
+                http_status_code=e.response.status_code,
+            ) from e
+
+        topic = TableflowTopic.from_response(response.json())
+        if wait_for_running:
+            return self._wait_for_tableflow_running(topic, timeout)
+        return topic
+
     def disable_tableflow(
         self,
         table_name: str,