afv-library/skills/data360-code-extension-generate/SKILL.md

name	description	metadata
data360-code-extension-generate	Develop and deploy Data Cloud Code Extensions using SF CLI plugin. Use this skill when creating custom Python transformations for Data Cloud, deploying code extensions, or testing data transformations. Supports init, run, scan, and deploy operations.	version relatedSkills cliTools 1.0 data360-schema-get tool semver docker >=20.0.0 tool semver pip >=23.0.0 tool semver python >=3.11.0 tool semver python3 >=3.11.0 tool semver sf >=2.0.0

data360-code-extension-generate Skill

Overview

This skill provides a complete workflow for developing, testing, and deploying custom Python code extensions to Salesforce Data Cloud. Code extensions allow you to write Python transformations that read from and write to Data Lake Objects (DLOs) and Data Model Objects (DMOs).

When to Use

• User wants to create a new code extension project
• User needs to test a code extension locally
• User wants to scan code for required permissions
• User needs to deploy a code extension to Data Cloud
• User wants to run a deployed code extension in the org
• User wants to query code extension logs in an org
• User is working with Data Cloud transformations
• User wants to read/write DLO or DMO data programmatically

Prerequisites Check

Before executing any code extension commands, verify prerequisites:

1. SF CLI with plugin installed

   sf plugins --core | grep data-code-extension
   

   If not installed:

   sf plugins install @salesforce/plugin-data-code-extension
   

2. Python 3.11

   python --version  # Should show 3.11.x
   

3. Data Cloud Custom Code SDK

   pip list | grep salesforce-data-customcode
   

   If not installed:

   pip install salesforce-data-customcode
   

4. Docker running (for deploy only)

   docker ps
   

   If Docker is not running, automatically launch it and wait for the daemon:

   open -a Docker && for i in $(seq 1 30); do docker ps > /dev/null 2>&1 && echo "Docker ready" && break || sleep 2; done
   

5. Authenticated org

   sf org display --target-org <org_alias> --json
   

Skill Workflow

Phase 1: Initialize Project

Create a new code extension project with scaffolding.

Commands:

For script-based code extensions (batch transformations):

sf data-code-extension script init --package-dir <directory>

For function-based code extensions (real-time):

sf data-code-extension function init --package-dir <directory>

Required Option:

• --package-dir, -p - Directory path where the package will be created

What it creates:

my-transform/              # Project root
├── payload/               # CRITICAL: This is what --package-dir must point to for deploy
│   ├── entrypoint.py      # Main transformation code
│   └── config.json        # Code extension configuration
├── requirements.txt       # Python dependencies
└── README.md

Directory Context During Workflow

IMPORTANT: Understanding the directory structure is critical for successful deployment.

Commands and their directory requirements:

Command	Run From	Path/File Argument
init	Parent directory	<project-name> or .
scan	Project root	./payload/entrypoint.py
run	Project root	./payload/entrypoint.py
deploy	Project root	--package-dir ./payload (REQUIRED)

CRITICAL: The --package-dir argument in deploy command MUST point to the payload directory, not the project root.

Phase 2: Develop Transformation

Edit payload/entrypoint.py with transformation logic.

Script Example (Batch):

client.read_dlo() returns a PySpark DataFrame, not Pandas. Data Cloud's sandboxed Spark environment blocks certain operations (UDFs, rdd.map(), Arrow config). The safest pattern for row-level Python transformations is .toPandas() → transform → spark.createDataFrame():

from pyspark.sql import SparkSession

from datacustomcode.client import Client
from datacustomcode.io.writer.base import WriteMode

def main():
    client = Client()

    # Read from DLO (returns PySpark DataFrame)
    df = client.read_dlo('Employee__dll')

    # Convert to Pandas, transform, convert back
    pdf = df.toPandas()
    pdf['position__c'] = pdf['position__c'].str.upper()
    spark = SparkSession.builder.getOrCreate()
    df_result = spark.createDataFrame(pdf, schema=df.schema)

    # Write to output DLO
    client.write_to_dlo('Employee_Upper__dll', df_result, write_mode=WriteMode.APPEND)

if __name__ == "__main__":
    main()

For PySpark-native column operations that don't require Python UDFs (e.g., withColumn, filter, select), you can operate directly on the DataFrame without converting to Pandas.

Function Example (Real-time):

from datacustomcode.function_client import FunctionClient

def transform(event, context):
    client = FunctionClient(context)
    input_data = event['data']
    output = {
        'name': input_data['name'].upper(),
        'status': 'processed'
    }
    return output

Common Operations:

• client.read_dlo('DLO_Name__dll') - Read from DLO (returns PySpark DataFrame)
• client.read_dmo('DMO_Name') - Read from DMO (returns PySpark DataFrame)
• client.write_to_dlo('DLO_Name__dll', df, write_mode=WriteMode.APPEND) - Write to DLO
• client.write_to_dlo('DLO_Name__dll', df, write_mode=WriteMode.OVERWRITE) - Overwrite DLO
• client.write_to_dmo('DMO_Name', df, write_mode=WriteMode.UPSERT) - Upsert to DMO

Sandbox Limitations (Data Cloud Spark Environment):

• spark.sql.execution.pythonUDF.arrow.enabled config is blocked — PySpark UDFs will fail
• rdd.map() is blocked — cannot use RDD-level Python transformations
• Use .toPandas() for row-level Python logic, then convert back with spark.createDataFrame()
• PySpark-native column operations (withColumn, filter, select, SQL expressions) work fine without conversion

Phase 3: Scan for Permissions

Scan the entrypoint file to detect required permissions and generate config.json.

Command:

sf data-code-extension script scan --entrypoint ./payload/entrypoint.py

What it detects:

• Read permissions for DLOs/DMOs
• Write permissions for DLOs/DMOs
• Python package dependencies
• Updates config.json and requirements.txt

Phase 4: Validate DLO Schema (Pre-Test Check)

CRITICAL: Before running tests locally, validate that all DLOs used in your code exist and have the expected fields.

Step 4a: Extract DLOs from config.json

After scanning, review the generated config.json to identify all DLOs:

cat payload/config.json

Step 4b: Validate Each DLO Schema

Use the data360-schema-get skill to verify DLOs exist and check field names.

For each DLO referenced in your code:

1. Verify DLO exists:

   python3 scripts/get_dlo_schema.py <org_alias> <dlo_name>
   

2. Verify field names match — compare fields used in your entrypoint.py against the DLO schema.

3. Check all DLOs:

   • Validate all DLOs in read permissions
   • Validate all DLOs in write permissions
   • Check field names match exactly (case-sensitive)
   • Verify data types are compatible with operations

Step 4c: Validation Checklist

Before proceeding to run, ensure:

• All DLOs in config.json exist in target org
• All field names used in code exist in DLO schemas
• Field data types match your transformation logic
• Primary key fields are correctly identified
• Write target DLOs are created and accessible

Phase 5: Run and Test Locally

After validating DLO schemas, run the code extension locally against your Data Cloud org.

If you arrived here because you were asked to run a code extension in the org, skip to phases 7a and 7b.

Step 5a: Determine Target Org

Infer the default org by running:

sf config get target-org --json

If a default org is found, present it to the user for confirmation rather than asking them to provide one from scratch.

Step 5b: Run the Code Extension

For script code extensions:

sf data-code-extension script run --entrypoint <entrypoint_file> --target-org <org_alias> [options]

For function code extensions (with model callouts):

sf data-code-extension function run --entrypoint <entrypoint_file> --test-with <test_json> --target-org <org_alias>

For function code extensions (no model callouts):

sf data-code-extension function run --entrypoint <entrypoint_file> --test-with <test_json>

Options:

• --target-org, -o - SF CLI org alias (required for scripts only)
• --test-with, -t - Path to test.json input (required for functions only)
• --config-file, -c - Custom config file path

If you get errors:

• Re-validate DLO schemas
• Check field names are exact matches
• Verify data types are compatible
• Review error messages for field/DLO issues

Phase 6: Deploy to Data Cloud

Deploy the code extension to Data Cloud for scheduled or on-demand execution.

CRITICAL: You MUST specify --package-dir ./payload to point to the payload directory created by init.

Command:

sf data-code-extension script deploy --target-org <org_alias> --name <name> --package-dir ./payload --package-version <version> --description <description> [options]

Required Options:

• --target-org, -o - SF CLI org alias
• --name, -n - Name for code extension deployment
• --package-dir - Path to payload directory (REQUIRED - must be ./payload when running from project root)
• --package-version - Version string (default: 0.0.1)
• --description - Description of code extension

Optional Options:

• --cpu-size - CPU size: CPU_L, CPU_XL, CPU_2XL (default), CPU_4XL
• --function-invoke-opt - Function invoke options (for function type)
• --network - Docker network (default: default)

After this phase, choose 7a or 7b, depending if it's a script or a function.

Phase 7a: Run a Batch Transform Script in the Org (needs Data 360 MCP server - https://github.com/forcedotcom/d360-mcp-server).

• A deployed code extension script creates a matching Data Transform in the org
• The transform status will initially be PROCESSING — wait ~30 seconds and poll d360_transform_get until it becomes ACTIVE before attempting running it
• Continue with the next instructions:

Step 7a.1: Find the Transform

Use d360_transform_list to find the deployed code extension by name:

mcp__data360__execute(toolName="d360_transform_list", paramsJson="{}")

Look for the transform matching your deployment name (the --name value from deploy). It will have "creationSource": "Code Extension" in the response.

Step 7a.2: Run the Transform

Use d360_transform_run with the transform name (the name field, not label):

mcp__data360__execute(toolName="d360_transform_run", paramsJson="{\"transformId\":\"<transform_name>\"}")

A successful response returns {"success": true}.

Note: After triggering a run, the transform status will change to PROCESSING. Poll with d360_transform_get to check lastRunStatus for completion. If the status was PROCESSING right after deploy, wait for ACTIVE before running.

Step 7a.3: Verify the Output

Once the run completes successfully (lastRunStatus: "SUCCESS"), query the target DLO to confirm data was written:

SELECT * FROM <target_DLO>__dll LIMIT 10

Use d360_query_sql to execute the query. Check that:

• Records exist in the target DLO
• The transformed fields contain expected values
• Row count matches expectations (compare with source DLO if applicable)

Phase 7b: Use a Chunking Function in a Search Index (needs Data 360 MCP server - https://github.com/forcedotcom/d360-mcp-server).

A deployed function-based code extension can be used as a custom chunking strategy in a Data Cloud Search Index. The function runs automatically each time the search index processes data — there's no manual "run" step.

Step 7b.1: Verify the Function is Deployed and Active

Use d360_transform_list or check the code extension list to confirm the function is deployed. Note the exact deployment name (e.g., text_chunking_v2) — this is the function_name value you'll use.

Step 7b.2: Create a Search Index with Custom Code Chunking

Use d360_search_index_create. The key difference from standard search indexes is the chunking configuration — use strategy custom_code with a function_name parameter pointing to your deployed function:

"chunkingConfiguration": {
  "fieldLevelConfigurations": [
    {
      "sourceDmoDeveloperName": "YourSource__dlm",
      "sourceDmoFieldDeveloperName": "your_text_field__c",
      "config": {
        "id": "custom_code",
        "userValues": [
          {
            "id": "function_name",
            "value": "<your_deployed_function_name>"
          }
        ]
      }
    }
  ]
}

The function_name value must exactly match the --name used during sf data-code-extension function deploy.

Step 7b.3: The Function Runs Automatically

Unlike batch script transforms (Phase 7a), function-based code extensions used in search indexes do not need to be triggered manually. The search index processing pipeline invokes the function automatically when:

• The index is first created (processes existing data)
• New data arrives in the source DMO (if processingType is NEAR_REALTIME)

Step 7b.4: Verify the Index Built Successfully

Check the search index status:

mcp__data360__execute(toolName="d360_search_index_get", paramsJson="{\"developerName\":\"<index_name>\"}")

Look for activationStatus: "ACTIVE" and check d360_search_index_process_history for build completion details.

Step 7b.5: Verify Chunks

Query the chunk DMO to confirm your function produced the expected output:

SELECT chunk_text__c, seq_no__c 
FROM <ChunkDMO>__dlm 
LIMIT 10

Phase 8: Monitor Logs (needs Data 360 MCP server - https://github.com/forcedotcom/d360-mcp-server).

Code extension print() output, when it runs in an org, is stored in DataCustomCodeLogs__dll. Query logs after a run:

SELECT Message__c, Timestamp__c 
FROM DataCustomCodeLogs__dll 
WHERE ProcessDefinitionName__c = '<transform_name>' 
ORDER BY Timestamp__c DESC 
LIMIT 50

Key fields:

• ExecutionId__c — unique ID per code extension execution
• DataCustomCodeName__c — the code extension name
• ProcessDefinitionName__c — the transform that triggered the run
• Message__c — the log message (from print() statements)
• Timestamp__c — when the log was emitted

To get logs for a specific run, filter by ExecutionId__c. To find the latest execution:

SELECT DISTINCT ExecutionId__c, MAX(Timestamp__c) as last_log
FROM DataCustomCodeLogs__dll
WHERE ProcessDefinitionName__c = '<transform_name>'
GROUP BY ExecutionId__c
ORDER BY last_log DESC
LIMIT 1

Error Handling

Common Issues and Solutions

Error	Solution
command data-code-extension not found	sf plugins install @salesforce/plugin-data-code-extension
datacustomcode CLI not found	pip install salesforce-data-customcode
Python version mismatch	Use pyenv: pyenv install 3.11.0 && pyenv local 3.11.0
Cannot connect to Docker daemon	Start Docker Desktop
No org found for alias	sf org login web --alias <org_alias>
config.json not found	sf data-code-extension script scan --entrypoint ./payload/entrypoint.py
DLO not found	Verify DLO exists (use data360-schema-get skill), check spelling and __dll suffix
Permission denied writing	Re-run scan, verify target DLO exists and is writable
Deploy fails - wrong directory	Ensure --package-dir points to payload/ directory, not project root
spark.sql.execution.pythonUDF.arrow.enabled permission denied	Sandbox blocks UDFs/Arrow. Use .toPandas() → transform → spark.createDataFrame() pattern instead
rdd.map() fails or transform shows FAILURE	RDD operations blocked in sandbox. Use .toPandas() pattern instead
Transform status is PROCESSING after deploy	Wait ~30 seconds, then poll d360_transform_get until status is ACTIVE before running

Best Practices

Development

1. Always scan before testing — run scan after code changes
2. Test locally first — use run command before deploying
3. Use version control — git commit after each successful test
4. Version your deployments — use semantic versioning (1.0.0, 1.1.0, etc.)
5. Deploy from project root with --package-dir ./payload

Performance

• CPU_L: Small datasets (< 1M records)
• CPU_2XL: Medium datasets (1M-10M records)
• CPU_4XL: Large datasets (> 10M records)

Security

1. No hardcoded credentials — use SF CLI authentication only
2. Validate input data — check for nulls and data types
3. Limit write permissions — only grant necessary DLO/DMO access

Integration with Other Skills

Use with data360-schema-get skill (CRITICAL for validation):

The data360-schema-get skill is required for validating DLOs before testing code extensions.

Use with Datakit Workflow:

1. Create DLO via code extension
2. Map DLO to DMO using datakit workflow
3. Use DMO in segments and activations

Command Reference

Command	Purpose	Required Args
script init	Create new script project	--package-dir
function init	Create new function project	--package-dir
script scan	Generate config	entrypoint file
function scan	Generate config	entrypoint file
script run	Test script locally	entrypoint file, --target-org
function run	Test function locally	entrypoint file, --test-with (--target-org only if using models)
script deploy	Deploy to Data Cloud	--target-org, --name, --package-dir, --package-version, --description

Resources

• SF CLI Plugin: https://github.com/salesforcecli/plugin-data-code-extension
• Python SDK: https://github.com/forcedotcom/datacloud-customcode-python-sdk
• Data Cloud Docs: https://help.salesforce.com/s/articleView?id=sf.c360_a_intro.htm
• Python SDK PyPI: https://pypi.org/project/salesforce-data-customcode/

Notes

• Code extensions run in isolated Python 3.11 environment
• Docker is required only for deployment, not for local testing
• Use SF CLI authentication only (no separate credential files)
• Scan command auto-detects permissions from code
• Local run uses actual Data Cloud data (not mocked)
• Deployments are versioned and can be rolled back in UI