Add write_dmo_deltas to StreamingClient

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 6.2.0
+
+### Added
+
+- **`StreamingClient.write_dmo_deltas(name, dataframe)`** – start a streaming query that writes each micro-batch to the target DMO and return the `StreamingQuery` handle. It is the DMO counterpart to `write_dlo_deltas` and pairs with `read_dmo_deltas()`.
+
 ## 6.1.0
 
 ### Added

README.md

@@ -158,6 +158,7 @@ For a streaming (delta) transform, use `StreamingClient`, which exposes the stre
 * `read_dlo_deltas()` – Read the streaming change feed (deltas) of a Data Lake Object as a streaming DataFrame.
 * `read_dmo_deltas()` – Read the streaming change feed (deltas) of a Data Model Object as a streaming DataFrame.
 * `write_dlo_deltas(name, spark_dataframe)` – Write a streaming DataFrame of deltas to a Data Lake Object; returns the started `StreamingQuery`
+* `write_dmo_deltas(name, spark_dataframe)` – Write a streaming DataFrame of deltas to a Data Model Object; returns the started `StreamingQuery`
 
 `find_file_path`, `llm_gateway_generate_text`, and `einstein_predict` are available on both clients.
 
@@ -222,6 +223,15 @@ query = client.write_dlo_deltas("Output__dll", transformed)
 query.awaitTermination()
 ```
 
+A streaming transform can also read and write Data Model Objects with `read_dmo_deltas()` and `write_dmo_deltas()`:
+
+```python
+deltas = client.read_dmo_deltas()
+transformed = deltas.withColumn("description__c", upper(col("description__c")))
+query = client.write_dmo_deltas("Output__dlm", transformed)
+query.awaitTermination()
+```
+
 Notes:
 
 - These methods only run inside the Data Cloud streaming (`DELTA_SYNC`) runtime. Locally (`datacustomcode run`) they raise `NotImplementedError`, since there is no change feed to stream.

src/datacustomcode/client.py

@@ -726,6 +726,25 @@ def write_dlo_deltas(
         self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DMO)
         return self._writer.write_dlo_deltas(name, dataframe, **kwargs)  # type: ignore[no-any-return]
 
+    def write_dmo_deltas(
+        self, name: str, dataframe: PySparkDataFrame, **kwargs
+    ) -> StreamingQuery:
+        """Write a streaming DataFrame of deltas to a DMO in Data Cloud.
+
+        Starts a streaming query that writes each micro-batch to the
+        target DMO and returns the ``StreamingQuery`` handle; the caller
+        typically calls ``query.awaitTermination()``.
+
+        Args:
+            name: The name of the DMO to write to.
+            dataframe: The streaming PySpark DataFrame to write.
+
+        Returns:
+            The started ``StreamingQuery``.
+        """
+        self._validate_data_layer_history_does_not_contain(DataCloudObjectType.DLO)
+        return self._writer.write_dmo_deltas(name, dataframe, **kwargs)  # type: ignore[no-any-return]
+
     def auto_write_to_dlo(self, name: str, dataframe: PySparkDataFrame) -> None:
         """Write a PySpark DataFrame to a DLO in Data Cloud automatically picking
         the WriteMode.

src/datacustomcode/deploy.py

@@ -415,13 +415,7 @@ def is_streaming(self) -> bool:
     def _validate_layers(self) -> "DataTransformConfig":
         read_is_dlo = isinstance(self.permissions.read, DloPermission)
         write_is_dlo = isinstance(self.permissions.write, DloPermission)
-        if self.is_streaming:
-            if not write_is_dlo:
-                raise ValueError(
-                    "A streaming transform must write to a DLO "
-                    "(permissions.write must be a 'dlo' entry)."
-                )
-        elif read_is_dlo != write_is_dlo:
+        if read_is_dlo != write_is_dlo:
             raise ValueError(
                 "permissions.read and permissions.write must both reference "
                 "DLOs or both reference DMOs (got "

src/datacustomcode/io/writer/base.py

@@ -87,7 +87,7 @@ def write_dlo_deltas(
         Args:
             name: Target Data Lake Object name.
             dataframe: Streaming PySpark DataFrame produced from a
-                ``read_dlo_deltas`` / ``read_dmo_deltas`` source.
+                ``read_dlo_deltas`` source.
 
         Returns:
             The started ``StreamingQuery``; the caller drives its lifecycle
@@ -102,3 +102,35 @@ def write_dlo_deltas(
             "streaming runtime; the local writer does not support streaming "
             "deltas."
         )
+
+    def write_dmo_deltas(
+        self, name: str, dataframe: PySparkDataFrame
+    ) -> StreamingQuery:
+        """Write a streaming DataFrame of deltas to a Data Model Object.
+
+        Streaming counterpart to :meth:`write_to_dmo`. Starts a streaming query
+        that writes each micro-batch to the target DMO via the Data Cloud
+        streaming sink and returns the resulting ``StreamingQuery`` handle. The
+        runtime owns the trigger and checkpoint location; callers pass only the
+        table name. Concrete streaming behavior is provided by the deployed
+        Data Cloud runtime; the base implementation raises
+        :class:`NotImplementedError`.
+
+        Args:
+            name: Target Data Model Object name.
+            dataframe: Streaming PySpark DataFrame produced from a
+                ``read_dmo_deltas`` source.
+
+        Returns:
+            The started ``StreamingQuery``; the caller drives its lifecycle
+            (typically ``query.awaitTermination()``).
+
+        Raises:
+            NotImplementedError: If the active writer does not support streaming
+                deltas (e.g. the local development writers).
+        """
+        raise NotImplementedError(
+            "write_dmo_deltas is only supported when running in the Data Cloud "
+            "streaming runtime; the local writer does not support streaming "
+            "deltas."
+        )

src/datacustomcode/scan.py

@@ -34,7 +34,7 @@
 
 DATA_ACCESS_METHODS = ["read_dlo", "read_dmo", "write_to_dlo", "write_to_dmo"]
 STREAMING_READ_METHODS = ["read_dlo_deltas", "read_dmo_deltas"]
-STREAMING_WRITE_METHODS = ["write_dlo_deltas"]
+STREAMING_WRITE_METHODS = ["write_dlo_deltas", "write_dmo_deltas"]
 
 DATA_TRANSFORM_CONFIG_TEMPLATE = {
     "sdkVersion": get_version(),
@@ -181,6 +181,7 @@ class StreamingDataAccessLayerCalls(pydantic.BaseModel):
     read_dlo_deltas: bool
     read_dmo_deltas: bool
     write_dlo_deltas: frozenset[str]
+    write_dmo_deltas: frozenset[str]
 
     @pydantic.model_validator(mode="after")
     def validate_access_layer(self) -> StreamingDataAccessLayerCalls:
@@ -193,10 +194,19 @@ def validate_access_layer(self) -> StreamingDataAccessLayerCalls:
                 "A streaming transform must read from at least one DLO or DMO "
                 "delta stream (read_dlo_deltas / read_dmo_deltas)."
             )
-        if not self.write_dlo_deltas:
+        if not self.write_dlo_deltas and not self.write_dmo_deltas:
             raise ValueError(
-                "A streaming transform must write to at least one DLO via "
-                "write_dlo_deltas."
+                "A streaming transform must write to at least one DLO or DMO "
+                "via write_dlo_deltas / write_dmo_deltas."
+            )
+        if self.write_dlo_deltas and self.write_dmo_deltas:
+            raise ValueError(
+                "Cannot write DLO and DMO deltas in the same streaming transform."
+            )
+        if self.read_layer != self.write_layer:
+            raise ValueError(
+                "A streaming transform must read and write the same layer "
+                f"(got read={self.read_layer}, write={self.write_layer})."
             )
         return self
 
@@ -205,6 +215,11 @@ def read_layer(self) -> str:
         """Return the read source layer, ``"dlo"`` or ``"dmo"``."""
         return "dlo" if self.read_dlo_deltas else "dmo"
 
+    @property
+    def write_layer(self) -> str:
+        """Return the write target layer, ``"dlo"`` or ``"dmo"``."""
+        return "dlo" if self.write_dlo_deltas else "dmo"
+
 
 class ClientMethodVisitor(ast.NodeVisitor):
     """AST Visitor that finds all instances of Client read/write method calls."""
@@ -217,6 +232,7 @@ def __init__(self) -> None:
         self._read_dlo_deltas: bool = False
         self._read_dmo_deltas: bool = False
         self._write_dlo_deltas_instances: set[str] = set()
+        self._write_dmo_deltas_instances: set[str] = set()
         self.variable_values: Dict[str, Union[str, None]] = {}
 
     def visit_Assign(self, node: ast.Assign) -> None:
@@ -261,6 +277,8 @@ def visit_Call(self, node: ast.Call) -> None:
                 name = self._resolve_name_arg(node.args[0])
                 if name and method_name == "write_dlo_deltas":
                     self._write_dlo_deltas_instances.add(name)
+                elif name and method_name == "write_dmo_deltas":
+                    self._write_dmo_deltas_instances.add(name)
         self.generic_visit(node)
 
     def _resolve_name_arg(self, arg: ast.expr) -> Union[str, None]:
@@ -278,6 +296,7 @@ def is_streaming(self) -> bool:
             self._read_dlo_deltas
             or self._read_dmo_deltas
             or bool(self._write_dlo_deltas_instances)
+            or bool(self._write_dmo_deltas_instances)
         )
 
     def found(self) -> DataAccessLayerCalls:
@@ -293,6 +312,7 @@ def found_streaming(self) -> StreamingDataAccessLayerCalls:
             read_dlo_deltas=self._read_dlo_deltas,
             read_dmo_deltas=self._read_dmo_deltas,
             write_dlo_deltas=frozenset(self._write_dlo_deltas_instances),
+            write_dmo_deltas=frozenset(self._write_dmo_deltas_instances),
         )
 
 
@@ -516,10 +536,10 @@ def _update_streaming_config(existing_config: dict[str, Any], file_path: str) ->
         )
 
     read_names = [source_name] if source_name else []
-    write_names = list(output.write_dlo_deltas)
+    write_names = list(output.write_dlo_deltas or output.write_dmo_deltas)
     existing_config["permissions"] = {
         "read": {read_layer: read_names},
-        "write": {"dlo": write_names},
+        "write": {output.write_layer: write_names},
     }
 
 

src/datacustomcode/templates/function/example/chunking_with_external_callout/README.md

@@ -0,0 +1,120 @@
+# Chunking with a Gemini Named Credential Callout
+
+Splits each input document into paragraph-sized chunks and calls Google's
+**Gemini** `generateContent` API for every chunk. The model returns a summary,
+category, sentiment, and topics, which are attached to the chunk as citations so
+the search index can filter and rank on them. Gemini is reached through a
+**Named Credential**, so this code never handles the endpoint URL or the API key.
+
+## How the callout works
+
+```python
+CALLOUT_URL = "callout:gemini"  # callout:<NC name>[/<path>]
+
+request = (
+    HTTPRequestBuilder()
+    .set_url(CALLOUT_URL)
+    .set_method(HTTPMethod.POST)
+    .set_headers({"Content-Type": "application/json"})
+    .set_response_timeout_seconds(60) # optional: per-callout timeout (seconds)
+    .build()
+)
+# Body is sent verbatim (serialize it yourself); the response body is a raw string.
+response = runtime.named_credential.request(request, json.dumps(payload))
+if response.is_success:
+    envelope = json.loads(response.body)
+    text = envelope["candidates"][0]["content"]["parts"][0]["text"]
+```
+
+The request asks for `responseMimeType: application/json` with a `responseSchema`,
+so Gemini returns the classification as a JSON string in
+`candidates[0].content.parts[0].text` — decode it, then decode that text again.
+
+The `gemini` Named Credential's URL already includes the full
+`/v1beta/models/<model>:generateContent` path, so the callout is just
+`callout:gemini` with **no path suffix** (anything after the name is appended to
+the credential's URL).
+
+## Configure the Named Credential
+
+1. Create an **External Credential** (e.g. `google_api_key`) that injects your
+   Gemini API key as the `X-goog-api-key` header.
+2. Create a **Named Credential** named `gemini`:
+   - **URL**: `https://generativelanguage.googleapis.com/v1beta/models/gemini-flash-latest:generateContent`
+   - **Enabled for Callouts** + **Generate Authorization Header**: on
+   - **External Credential**: `google_api_key`
+
+## Test locally
+
+```bash
+DATACUSTOMCODE_EXTERNAL_CALLOUT_CONFIG=/abs/path/to/external_callout_config.json \
+  sf data-code-extension function run \
+    --entrypoint payload/entrypoint.py \
+    --test-with payload/tests/test.json \
+    --target-org <your-org-alias>
+```
+
+With `--target-org` the SDK fetches only the **URL** from the org's Named
+Credential; **auth is always taken from `external_callout_config.json`** locally
+(the org's External Credential is used only in the Data Cloud runtime). So the
+`X-goog-api-key` must be in the local config for a local test. Omit
+`--target-org` to run fully offline using `target_url`.
+
+```json
+{
+    "credentials": {
+        "callout:gemini": {
+            "auth_type": "Custom",
+            "custom_headers": { "X-goog-api-key": "YOUR_GEMINI_API_KEY" },
+            "target_url": "https://generativelanguage.googleapis.com/v1beta/models/gemini-flash-latest:generateContent"
+        }
+    }
+}
+```
+
+Place `external_callout_config.json` in the **parent of your payload folder** (or
+point `DATACUSTOMCODE_EXTERNAL_CALLOUT_CONFIG` at it). It is never packaged into
+the deployment zip. Get a key from [Google AI Studio](https://aistudio.google.com/apikey);
+**do not commit it.**
+
+## Auth types
+
+`auth_type` selects how auth is injected for local testing. It should mirror the
+External Credential your Named Credential uses in the org, so local and deployed
+runs behave the same. This example uses `Custom` (Gemini's `X-goog-api-key`);
+all four supported types:
+
+```json
+{
+    "credentials": {
+        "callout:my_custom_api": {
+            "auth_type": "Custom",
+            "custom_headers": { "X-goog-api-key": "YOUR_API_KEY" }
+        },
+        "callout:my_basic_api": {
+            "auth_type": "Basic",
+            "username": "svc_user",
+            "password": "YOUR_PASSWORD"
+        },
+        "callout:my_oauth_api": {
+            "auth_type": "OAuth",
+            "access_token": "YOUR_ACCESS_TOKEN"
+        },
+        "callout:my_jwt_api": {
+            "auth_type": "Jwt",
+            "token": "YOUR_JWT"
+        }
+    }
+}
+```
+
+| `auth_type` | Fields read | Header sent |
+| ----------- | ------------------------------- | --------------------------------------- |
+| `Basic`     | `username`, `password`          | `Authorization: Basic <base64 user:pw>` |
+| `Custom`    | `custom_headers` (sent verbatim)| the headers you list                    |
+| `OAuth`     | `access_token` or `token`       | `Authorization: Bearer <token>`         |
+| `Jwt`       | `access_token` or `token`       | `Authorization: Bearer <token>`         |
+
+`OAuth`/`Jwt` take a token you supply for the local run — the SDK does not fetch
+or refresh it. In the Data Cloud runtime the Named Credential handles token
+acquisition; this local config only stands in for that during testing.

src/datacustomcode/templates/function/example/chunking_with_external_callout/config.json

@@ -0,0 +1,3 @@
+{
+    "entryPoint": "entrypoint.py"
+}