Support creating and deploying streaming script code packages

src/datacustomcode/cli.py

@@ -283,20 +283,43 @@ def deploy(
 )
 @click.option(
     "--use-in-feature",
-    default="SearchIndexChunking",
-    help="Feature where this function will be used (only applicable for function).",
+    "-u",
+    default=None,
+    help=(
+        "Invoke option for this package. For scripts: 'BatchTransform' "
+        "(default) or 'StreamingTransform'. For functions: 'SearchIndexChunking'."
+    ),
 )
 def init(directory: str, code_type: str, use_in_feature: Optional[str]):
+    from datacustomcode.constants import (
+        SCRIPT_USE_IN_FEATURE_BATCH,
+        SCRIPT_USE_IN_FEATURE_OPTIONS,
+        SCRIPT_USE_IN_FEATURE_STREAMING,
+    )
     from datacustomcode.scan import (
         dc_config_json_from_file,
         update_config,
         write_sdk_config,
     )
     from datacustomcode.template import copy_function_template, copy_script_template
 
+    streaming = False
+    if code_type == "script":
+        use_in_feature = use_in_feature or SCRIPT_USE_IN_FEATURE_BATCH
+        if use_in_feature not in SCRIPT_USE_IN_FEATURE_OPTIONS:
+            click.secho(
+                f"Error: Invalid --use-in-feature '{use_in_feature}' for a "
+                f"script. Valid options: {', '.join(SCRIPT_USE_IN_FEATURE_OPTIONS)}.",
+                fg="red",
+            )
+            raise click.Abort()
+        streaming = use_in_feature == SCRIPT_USE_IN_FEATURE_STREAMING
+    else:
+        use_in_feature = use_in_feature or "SearchIndexChunking"
+
     click.echo("Copying template to " + click.style(directory, fg="blue", bold=True))
     if code_type == "script":
-        copy_script_template(directory)
+        copy_script_template(directory, streaming=streaming)
     elif code_type == "function":
         copy_function_template(directory, use_in_feature)
     entrypoint_path = os.path.join(directory, PAYLOAD_DIR, ENTRYPOINT_FILE)
@@ -306,7 +329,9 @@ def init(directory: str, code_type: str, use_in_feature: Optional[str]):
     sdk_config = {"type": code_type}
     write_sdk_config(directory, sdk_config)
 
-    config_json = dc_config_json_from_file(entrypoint_path, code_type)
+    config_json = dc_config_json_from_file(
+        entrypoint_path, code_type, streaming=streaming
+    )
     with open(config_location, "w") as f:
         json.dump(config_json, f, indent=2)
 

src/datacustomcode/constants.py

@@ -38,6 +38,14 @@
     "SearchIndexChunking": "UnstructuredChunking",
 }
 
+# Script (data transform) invoke options
+SCRIPT_USE_IN_FEATURE_BATCH = "BatchTransform"
+SCRIPT_USE_IN_FEATURE_STREAMING = "StreamingTransform"
+SCRIPT_USE_IN_FEATURE_OPTIONS = [
+    SCRIPT_USE_IN_FEATURE_BATCH,
+    SCRIPT_USE_IN_FEATURE_STREAMING,
+]
+
 # Pydantic request/response type names to feature names
 REQUEST_TYPE_TO_FEATURE = {
     "SearchIndexChunkingV1Request": "SearchIndexChunking",

src/datacustomcode/deploy.py

@@ -14,6 +14,7 @@
 # limitations under the License.
 from __future__ import annotations
 
+import copy
 from html import unescape
 import json
 import os
@@ -41,6 +42,7 @@
 
 DATA_CUSTOM_CODE_PATH = "services/data/v63.0/ssot/data-custom-code"
 DATA_TRANSFORMS_PATH = "services/data/v63.0/ssot/data-transforms"
+DATA_CUSTOM_CODE_INVOKE_OPTIONS_PATH = "services/data/v67.0/ssot/data-custom-code"
 WAIT_FOR_DEPLOYMENT_TIMEOUT = 3000
 
 # Available compute types for Data Cloud deployments.
@@ -108,6 +110,7 @@ class CodeExtensionMetadata(BaseModel):
     computeType: str
     codeType: str
     functionInvokeOptions: Union[list[str], None] = None
+    invokeOptions: Union[list[str], None] = None
 
     def __init__(self, **data):
         name = data.get("name", "")
@@ -200,7 +203,14 @@ def create_deployment(
     access_token: AccessTokenResponse, metadata: CodeExtensionMetadata
 ) -> CreateDeploymentResponse:
     """Create a custom code deployment in the DataCloud."""
-    url = _join_strip_url(access_token.instance_url, DATA_CUSTOM_CODE_PATH)
+    # invokeOptions only binds at v67.0; route there when it is set so the
+    # option isn't silently dropped. Everything else stays on v63.0.
+    code_custom_code_path = (
+        DATA_CUSTOM_CODE_INVOKE_OPTIONS_PATH
+        if metadata.invokeOptions
+        else DATA_CUSTOM_CODE_PATH
+    )
+    url = _join_strip_url(access_token.instance_url, code_custom_code_path)
     body = dict[str, Any](
         {
             "label": metadata.name,
@@ -213,6 +223,8 @@ def create_deployment(
     )
     if metadata.functionInvokeOptions:
         body["functionInvokeOptions"] = metadata.functionInvokeOptions
+    if metadata.invokeOptions:
+        body["invokeOptions"] = metadata.invokeOptions
     logger.debug(f"Creating deployment {metadata.name}...")
     try:
         response = _make_api_call(
@@ -388,6 +400,30 @@ class DataTransformConfig(BaseConfig):
     dataspace: str
     permissions: Permissions
     dataObjects: Optional[list[DataObject]] = None
+    streamingSource: Optional[StreamingSource] = None
+
+    @property
+    def is_streaming(self) -> bool:
+        return self.streamingSource is not None
+
+    @model_validator(mode="after")
+    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:
+            raise ValueError(
+                "permissions.read and permissions.write must both reference "
+                "DLOs or both reference DMOs (got "
+                f"read={type(self.permissions.read).__name__}, "
+                f"write={type(self.permissions.write).__name__})"
+            )
+        return self
 
 
 class FunctionConfig(BaseConfig):
@@ -402,23 +438,15 @@ class DmoPermission(BaseModel):
     dmo: list[str]
 
 
+class StreamingSource(BaseModel):
+    type: str
+    name: str
+
+
 class Permissions(BaseModel):
     read: Union[DloPermission, DmoPermission]
     write: Union[DloPermission, DmoPermission]
 
-    @model_validator(mode="after")
-    def _no_mixed_layers(self) -> "Permissions":
-        read_is_dlo = isinstance(self.read, DloPermission)
-        write_is_dlo = isinstance(self.write, DloPermission)
-        if read_is_dlo != write_is_dlo:
-            raise ValueError(
-                "permissions.read and permissions.write must both reference "
-                "DLOs or both reference DMOs (got "
-                f"read={type(self.read).__name__}, "
-                f"write={type(self.write).__name__})"
-            )
-        return self
-
 
 def _permission_entries(perm: Union[DloPermission, DmoPermission]) -> list[str]:
     """Return the list of object names regardless of layer (DLO or DMO)."""
@@ -490,7 +518,9 @@ def create_data_transform(
 ) -> dict:
     """Create a data transform in the DataCloud."""
     script_name = metadata.name
-    request_hydrated = DATA_TRANSFORM_REQUEST_TEMPLATE.copy()
+    # Deep copy: the template's nested nodes/sources/macros dicts would
+    # otherwise be shared across calls and accumulate entries between deploys.
+    request_hydrated = copy.deepcopy(DATA_TRANSFORM_REQUEST_TEMPLATE)
 
     # Add nodes for each write entry (DLO or DMO)
     for i, name in enumerate(
@@ -533,7 +563,7 @@ def create_data_transform(
         "definition": definition,
         "label": f"{metadata.name}",
         "name": f"{metadata.name}",
-        "type": "BATCH",
+        "type": "STREAMING" if data_transform_config.is_streaming else "BATCH",
         "dataSpaceName": data_transform_config.dataspace,
     }
 
@@ -616,9 +646,18 @@ def deploy_full(
     callback=None,
 ) -> AccessTokenResponse:
     """Deploy a data transform in the DataCloud."""
+    from datacustomcode.constants import SCRIPT_USE_IN_FEATURE_STREAMING
+
     # prepare payload
     config = get_config(directory)
 
+    if (
+        isinstance(config, DataTransformConfig)
+        and config.is_streaming
+        and not metadata.invokeOptions
+    ):
+        metadata.invokeOptions = [SCRIPT_USE_IN_FEATURE_STREAMING]
+
     # create deployment and upload payload
     deployment = create_deployment(access_token, metadata)
     zip(directory, docker_network, metadata.codeType)