d6e-docker-stf-development

Docker STFs are containerized applications that execute as workflow steps in D6E. They read JSON from stdin, process data with custom logic, access workspace databases via internal API, and output JSON to stdout.

Apply this skill when users request:

• "Create a D6E Docker STF that..."
• "Build a custom STF for D6E that..."
• "I need a Docker-based workflow step..."
• "Help me create a data processing function for D6E"

Input Format

Docker STFs receive this JSON via stdin:

```json

{

"workspace_id": "UUID",

"stf_id": "UUID",

"caller": "UUID | null",

"api_url": "http://api:8080",

"api_token": "internal_token",

"input": {

"operation": "...",

...user-defined parameters

},

"sources": {

"step_name": {

"output": {...previous step data}

}

}

}

```

Output Format

Success:

```json

{

"output": {

"status": "success",

...custom result data

}

}

```

Error:

```json

{

"error": "Error message",

"type": "ErrorType"

}

```

SQL API Access

Execute SQL via internal API:

Endpoint: POST /api/v1/workspaces/{workspace_id}/sql

Headers:

```

Authorization: Bearer {api_token}

X-Internal-Bypass: true

X-Workspace-ID: {workspace_id}

X-STF-ID: {stf_id}

```

Request:

```json

{ "sql": "SELECT * FROM my_table LIMIT 10" }

```

Restrictions:

• No DDL (CREATE, DROP, ALTER)
• Policy-controlled access
• Workspace scope only

Python Implementation

main.py:

```python

#!/usr/bin/env python3

import sys

import json

import requests

import logging

logging.basicConfig(stream=sys.stderr, level=logging.INFO)

def execute_sql(api_url, api_token, workspace_id, stf_id, sql):

"""Execute SQL via D6E internal API"""

url = f"{api_url}/api/v1/workspaces/{workspace_id}/sql"

headers = {

"Authorization": f"Bearer {api_token}",

"X-Internal-Bypass": "true",

"X-Workspace-ID": workspace_id,

"X-STF-ID": stf_id,

"Content-Type": "application/json"

}

response = requests.post(url, json={"sql": sql}, headers=headers)

response.raise_for_status()

return response.json()

def main():

try:

input_data = json.load(sys.stdin)

user_input = input_data["input"]

# Your business logic here

result = {"status": "success", "message": "Processed"}

print(json.dumps({"output": result}))

except Exception as e:

logging.error(f"Error: {str(e)}", exc_info=True)

print(json.dumps({"error": str(e), "type": type(e).__name__}))

sys.exit(1)

if __name__ == "__main__":

main()

```

Dockerfile:

```dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY main.py .

RUN chmod +x main.py

ENTRYPOINT ["python3", "main.py"]

```

requirements.txt:

```

requests>=2.31.0

```

Node.js Implementation

index.js:

```javascript

const axios = require("axios");

async function executeSql(apiUrl, apiToken, workspaceId, stfId, sql) {

const response = await axios.post(

${apiUrl}/api/v1/workspaces/${workspaceId}/sql,

{ sql },

{

headers: {

Authorization: Bearer ${apiToken},

"X-Internal-Bypass": "true",

"X-Workspace-ID": workspaceId,

"X-STF-ID": stfId,

"Content-Type": "application/json",

},

}

);

return response.data;

}

async function main() {

try {

const input = await readStdin();

const data = JSON.parse(input);

// Your business logic here

const result = { status: "success", message: "Processed" };

console.log(JSON.stringify({ output: result }));

} catch (error) {

console.error("Error:", error.message);

console.log(

JSON.stringify({

error: error.message,

type: error.name,

})

);

process.exit(1);

}

}

function readStdin() {

return new Promise((resolve) => {

let data = "";

process.stdin.on("data", (chunk) => (data += chunk));

process.stdin.on("end", () => resolve(data));

});

}

main();

```

Dockerfile:

```dockerfile

FROM node:18-slim

WORKDIR /app

COPY package*.json ./

RUN npm ci --omit=dev

COPY index.js .

ENTRYPOINT ["node", "index.js"]

```

When creating a Docker STF, ensure:

• [ ] Reads JSON from stdin
• [ ] Outputs JSON to stdout ({"output": {...}})
• [ ] Logs to stderr (stdout is for results only)
• [ ] Handles errors gracefully
• [ ] Uses small base images (e.g., python:3.11-slim)
• [ ] Includes error type in error responses
• [ ] Validates input parameters
• [ ] Uses environment variables for configuration

Security

• Never log sensitive data (tokens, passwords)
• Validate all user inputs
• Use parameterized SQL queries
• Keep dependencies up-to-date

Performance

• Use multi-stage builds to reduce image size
• Minimize dependencies
• Add .dockerignore to exclude unnecessary files
• Cache pip/npm installations

Error Handling

```python

try:

# Your logic

result = process_data(input_data)

print(json.dumps({"output": result}))

except ValueError as e:

# Validation errors

logging.error(f"Validation error: {str(e)}")

print(json.dumps({"error": str(e), "type": "ValidationError"}))

sys.exit(1)

except Exception as e:

# Unexpected errors

logging.error(f"Unexpected error: {str(e)}", exc_info=True)

print(json.dumps({"error": str(e), "type": type(e).__name__}))

sys.exit(1)

```

Logging

```python

import logging

# Log to stderr

logging.basicConfig(

stream=sys.stderr,

level=logging.INFO,

format='%(asctime)s - %(levelname)s - %(message)s'

)

logging.info("Processing started")

logging.debug(f"Input: {input_data}") # Detailed logs

logging.warning("Deprecated operation used")

logging.error("Failed to process", exc_info=True)

```

Data Validation Pattern

```python

def validate_input(user_input):

required_fields = ["operation", "table_name"]

for field in required_fields:

if field not in user_input:

raise ValueError(f"Missing required field: {field}")

if user_input["operation"] not in ["query", "insert", "update"]:

raise ValueError(f"Invalid operation: {user_input['operation']}")

return True

# Usage

try:

validate_input(input_data["input"])

except ValueError as e:

print(json.dumps({"error": str(e), "type": "ValidationError"}))

sys.exit(1)

```

Database Query Pattern

```python

def safe_query(api_context, table_name, filters):

"""Execute a safe parameterized query"""

# Build WHERE clause safely

where_conditions = []

for key, value in filters.items():

# Simple validation

if not key.isidentifier():

raise ValueError(f"Invalid column name: {key}")

where_conditions.append(f"{key} = '{value}'")

where_clause = " AND ".join(where_conditions) if where_conditions else "1=1"

sql = f"SELECT * FROM {table_name} WHERE {where_clause} LIMIT 100"

return execute_sql(

api_context["api_url"],

api_context["api_token"],

api_context["workspace_id"],

api_context["stf_id"],

sql

)

```

External API Pattern

```python

import requests

from requests.adapters import HTTPAdapter

from urllib3.util.retry import Retry

def create_session():

"""Create session with retry logic"""

session = requests.Session()

retry = Retry(

total=3,

backoff_factor=0.3,

status_forcelist=[500, 502, 503, 504]

)

adapter = HTTPAdapter(max_retries=retry)

session.mount('http://', adapter)

session.mount('https://', adapter)

return session

def call_external_api(url, params):

"""Call external API with error handling"""

session = create_session()

try:

response = session.get(url, params=params, timeout=10)

response.raise_for_status()

return response.json()

except requests.Timeout:

raise Exception("External API timeout")

except requests.RequestException as e:

raise Exception(f"External API error: {str(e)}")

```

Build and Test

```bash

# Build image

docker build -t my-stf:latest .

# Test with sample input

echo '{

"workspace_id": "test-id",

"stf_id": "test-stf-id",

"caller": null,

"api_url": "http://localhost:8080",

"api_token": "test-token",

"input": {

"operation": "test"

},

"sources": {}

}' | docker run --rm -i my-stf:latest

```

Debug Mode

```bash

# Run with interactive shell

docker run --rm -it --entrypoint /bin/bash my-stf:latest

# Check image size

docker images my-stf:latest

# Inspect logs

docker run --rm -i my-stf:latest < input.json 2>&1 | tee output.log

```

Issue: "Policy violation" error

Cause: STF doesn't have permission to access the table.

Solution: Create policies:

```javascript

// Create policy group

d6e_create_policy_group({ name: "my-stf-group" });

// Add STF to group

d6e_add_member_to_policy_group({

policy_group_id: "{group_id}",

member_type: "stf",

member_id: "{stf_id}",

});

// Grant access

d6e_create_policy({

policy_group_id: "{group_id}",

table_name: "my_table",

operation: "select",

mode: "allow",

});

```

Issue: Output not appearing in D6E

Cause: Output not in correct JSON format.

Solution: Always use {"output": {...}} format:

```python

# ✅ Correct

print(json.dumps({"output": {"status": "success"}}))

# ❌ Wrong

print(json.dumps({"status": "success"}))

```

Issue: "Image not found" in D6E

Cause: Image not accessible from D6E API server.

Solution:

1. Publish to container registry (GitHub, Docker Hub)
2. Or ensure same Docker daemon as D6E API server

Issue: Large image size

Solution: Use multi-stage builds:

```dockerfile

# Build stage

FROM python:3.11 AS builder

WORKDIR /app

COPY requirements.txt .

RUN pip install --user --no-cache-dir -r requirements.txt

# Runtime stage

FROM python:3.11-slim

WORKDIR /app

COPY --from=builder /root/.local /root/.local

COPY main.py .

ENV PATH=/root/.local/bin:$PATH

ENTRYPOINT ["python3", "main.py"]

```

Every Docker STF should include:

```

my-stf/

├── main.py (or index.js, main.go) # Entry point

├── Dockerfile # Container definition

├── requirements.txt (or package.json, go.mod) # Dependencies

├── .dockerignore # Exclude files

└── README.md # Documentation

```

.dockerignore:

```

.git

.gitignore

*.md

tests/

__pycache__/

*.pyc

node_modules/

.env

```

For detailed information:

• Complete API reference: [reference.md](reference.md)
• More implementation examples: [examples.md](examples.md)
• Quick start guide: [../docs/QUICKSTART.md](../../docs/QUICKSTART.md)
• Testing guide: [../docs/TESTING.md](../../docs/TESTING.md)
• Publishing guide: [../docs/PUBLISHING.md](../../docs/PUBLISHING.md)