Python Task Authoring¶
Execution Strategies¶
Python tasks typically use a bash wrapper with prelude.tmpl.
Strategy 1 — Bash Wrapper Writes a .py File Then Executes It¶
Templates:
- Name: "my-script.py"
Path: "/tmp/my-script.py"
Contents: |
#!/usr/bin/env python3
import sys, json, subprocess
# Python code here
- Name: "my-task.sh"
Contents: |
#!/usr/bin/env bash
{{ template "prelude.tmpl" . }}
python3 /tmp/my-script.py
exit $?
Strategy 2 — Inline Heredoc in Bash¶
Python Availability¶
- Ensure python3 is available (install if needed:
install python3) - On ESXi 7.x+, python3 is built-in
DRP Integration¶
Via Subprocess (drpcli)¶
import subprocess, json
result = subprocess.run(
["drpcli", "machines", "show", os.environ["RS_UUID"]],
capture_output=True, text=True
)
machine = json.loads(result.stdout)
Via Direct HTTP API¶
import requests, os
endpoint = os.environ["RS_ENDPOINT"]
token = os.environ["RS_TOKEN"]
headers = {"Authorization": f"Bearer {token}"}
resp = requests.get(f"{endpoint}/api/v3/machines/{os.environ['RS_UUID']}", headers=headers)
Parameter Passing¶
Parameters are passed via Go template rendering into the Python script.
Error Handling¶
- Use
sys.exit(0)for success,sys.exit(1)for failure - Use try/except for graceful error handling
- stdout is captured as job log — use
print()for logging - stderr is also captured — exceptions print to stderr by default
Recommended Pattern¶
try:
result = do_something()
print(f"INFO: Successfully completed: {result}")
sys.exit(0)
except Exception as e:
print(f"FATAL: {e}", file=sys.stderr)
sys.exit(1)
vSphere API Pattern¶
- Used by VMware tasks (esxi-no-vib-* family) for managing ESXi via vSphere API
- These run in a context (Linux host), NOT on ESXi directly
- Uses pyVmomi or direct SOAP/REST calls
- Example from esxi-no-vib-enable-ssh: Python template (.py.tmpl) that connects to vCenter/ESXi and enables SSH service
- Auth credentials typically from DRP params rendered into the script
Complete Example¶
---
Name: "example-python-api-check"
Description: "Check an external API endpoint using Python"
Documentation: |
Validates connectivity to an external API endpoint and records
the result as a machine parameter.
RequiredParams:
- example/api-endpoint
OptionalParams:
- example/api-timeout
- rs-debug-enable
Meta:
icon: "python"
color: "blue"
title: "Example Content"
feature-flags: "sane-exit-codes"
Templates:
- Name: "api-check.py"
Path: "/tmp/api-check.py"
Contents: |
#!/usr/bin/env python3
import sys, os, json, subprocess
endpoint = "{{ .Param "example/api-endpoint" }}"
{{ if .ParamExists "example/api-timeout" }}
timeout = {{ .Param "example/api-timeout" }}
{{ else }}
timeout = 30
{{ end }}
try:
import requests
resp = requests.get(endpoint, timeout=timeout)
resp.raise_for_status()
print(f"INFO: API check passed: {resp.status_code}")
# Store result back in DRP
subprocess.run([
"drpcli", "machines", "set",
os.environ["RS_UUID"], "param",
"example/api-status", "to", json.dumps({"status": "ok", "code": resp.status_code})
], check=True)
sys.exit(0)
except Exception as e:
print(f"FATAL: API check failed: {e}")
sys.exit(1)
- Name: "example-python-api-check.sh"
Meta:
OS: "linux"
Contents: |
#!/usr/bin/env bash
{{ template "prelude.tmpl" . }}
install python3 python3-pip
pip3 install requests 2>/dev/null || true
python3 /tmp/api-check.py
exit $?