Added invoke options for functions.

Author: sbyrne-sf

src/datacustomcode/cli.py

@@ -21,6 +21,8 @@
 import click
 from loguru import logger
 
+from datacustomcode.scan import find_base_directory, get_package_type
+
 
 @click.group()
 @click.option("--debug", is_flag=True)
@@ -160,6 +162,7 @@ def zip(path: str, network: str):
 
     Choose based on your workload requirements.""",
 )
+@click.option("--function-invoke-opt")
 def deploy(
     path: str,
     name: str,
@@ -168,6 +171,7 @@ def deploy(
     cpu_size: str,
     profile: str,
     network: str,
+    function_invoke_opt: str,
 ):
     from datacustomcode.credentials import Credentials
     from datacustomcode.deploy import CodeExtensionMetadata, deploy_full
@@ -193,6 +197,20 @@ def deploy(
         description=description,
         computeType=COMPUTE_TYPES[cpu_size],
     )
+
+    base_directory = find_base_directory(path)
+    package_type = get_package_type(base_directory)
+    if package_type == "function":
+        if not function_invoke_opt:
+            click.secho(
+                "Error: Function invoke options are required for function package type",
+                fg="red",
+            )
+            raise click.Abort()
+        else:
+            function_invoke_options = function_invoke_opt.split(",")
+            metadata.functionInvokeOptions = function_invoke_options
+
     try:
         credentials = Credentials.from_available(profile=profile)
     except ValueError as e:

src/datacustomcode/deploy.py

@@ -61,6 +61,7 @@ class CodeExtensionMetadata(BaseModel):
     version: str
     description: str
     computeType: str
+    functionInvokeOptions: Union[list[str], None] = None
 
     def __init__(self, **data):
         super().__init__(**data)
@@ -153,13 +154,17 @@ def create_deployment(
 ) -> CreateDeploymentResponse:
     """Create a custom code deployment in the DataCloud."""
     url = _join_strip_url(access_token.instance_url, DATA_CUSTOM_CODE_PATH)
-    body = {
-        "label": metadata.name,
-        "name": metadata.name,
-        "description": metadata.description,
-        "version": metadata.version,
-        "computeType": metadata.computeType,
-    }
+    body = dict[str, Any](
+        {
+            "label": metadata.name,
+            "name": metadata.name,
+            "description": metadata.description,
+            "version": metadata.version,
+            "computeType": metadata.computeType,
+        }
+    )
+    if metadata.functionInvokeOptions:
+        body["functionInvokeOptions"] = metadata.functionInvokeOptions
     logger.debug(f"Creating deployment {metadata.name}...")
     try:
         response = _make_api_call(
@@ -312,7 +317,7 @@ class DloPermission(BaseModel):
 
 
 def get_config(directory: str) -> BaseConfig:
-    """Get the data transform config from the config.json file."""
+    """Get the code extension config from the config.json file."""
     config_path = os.path.join(directory, "config.json")
     try:
         with open(config_path, "r") as f:

tests/test_cli.py

@@ -5,6 +5,7 @@
 from click.testing import CliRunner
 
 from datacustomcode.cli import deploy, init
+from datacustomcode.scan import write_sdk_config
 
 
 class TestInit:
@@ -95,6 +96,30 @@ def test_deploy_command_success(self, mock_credentials, mock_deploy_full):
             assert call_args[0][1].description == "Custom Data Transform Code"
             assert call_args[0][2] == mock_creds  # credentials
 
+    @patch("datacustomcode.deploy.deploy_full")
+    @patch("datacustomcode.credentials.Credentials.from_available")
+    def test_deploy_command_function_invoke_options(
+        self, mock_credentials, mock_deploy_full
+    ):
+        """Test deploy command with function invoke options."""
+        runner = CliRunner()
+        with runner.isolated_filesystem():
+            # Create test payload directory
+            os.makedirs("payload", exist_ok=True)
+            sdk_config = {"type": "function"}
+            write_sdk_config(".", sdk_config)
+            result = runner.invoke(
+                deploy,
+                ["--name", "test-job", "--function-invoke-opt", "option1,option2"],
+            )
+
+            assert result.exit_code == 0
+            mock_deploy_full.assert_called_once()
+
+            # Check that deploy_full was called with function invoke options
+            call_args = mock_deploy_full.call_args
+            assert call_args[0][1].functionInvokeOptions == ["option1", "option2"]
+
     @patch("datacustomcode.credentials.Credentials.from_available")
     def test_deploy_command_credentials_error(self, mock_credentials):
         """Test deploy command when credentials are not available."""

tests/test_deploy.py

@@ -531,6 +531,30 @@ def test_create_deployment_conflict(self, mock_make_api_call):
         with pytest.raises(ValueError, match="Deployment test_job exists"):
             create_deployment(access_token, metadata)
 
+    @patch("datacustomcode.deploy._make_api_call")
+    def test_create_deployment_function_invoke_options(self, mock_make_api_call):
+        """Test deployment creation with function invoke options."""
+        access_token = AccessTokenResponse(
+            access_token="test_token", instance_url="https://instance.example.com"
+        )
+        metadata = CodeExtensionMetadata(
+            name="test_job",
+            version="1.0.0",
+            description="Test job",
+            computeType="CPU_M",
+            functionInvokeOptions=["option1", "option2"],
+        )
+
+        mock_make_api_call.return_value = {
+            "fileUploadUrl": "https://upload.example.com"
+        }
+
+        result = create_deployment(access_token, metadata)
+
+        mock_make_api_call.assert_called_once()
+        assert isinstance(result, CreateDeploymentResponse)
+        assert result.fileUploadUrl == "https://upload.example.com"
+
 
 class TestZip:
     @patch("datacustomcode.deploy.has_nonempty_requirements_file")