from hopx_ai import Sandbox
import os
from typing import Dict, Any, List
class WorkflowNode:
def __init__(self, node_id: str, node_type: str, config: Dict[str, Any]):
self.node_id = node_id
self.node_type = node_type
self.config = config
class LowCodeBackend:
def __init__(self, api_key: str):
self.api_key = api_key
self.sandbox = None
def initialize(self):
"""Initialize backend"""
self.sandbox = Sandbox.create(
template="code-interpreter",
api_key=self.api_key,
timeout_seconds=3600
)
def translate_workflow(self, nodes: List[WorkflowNode]) -> str:
"""Translate visual workflow to code"""
code_parts = []
for node in nodes:
if node.node_type == "trigger":
code_parts.append(f"# Trigger: {node.config.get('name', '')}")
elif node.node_type == "action":
action_code = self._generate_action_code(node)
code_parts.append(action_code)
elif node.node_type == "condition":
condition_code = self._generate_condition_code(node)
code_parts.append(condition_code)
return "\n".join(code_parts)
def _generate_action_code(self, node: WorkflowNode) -> str:
"""Generate code for action node"""
action_type = node.config.get("action_type", "")
if action_type == "http_request":
return f"""
import requests
response = requests.{node.config.get('method', 'get').lower()}('{node.config.get('url', '')}')
print(response.text)
"""
elif action_type == "data_transform":
return f"""
data = {node.config.get('data', '{}')}
# Transform data
result = data
print(result)
"""
else:
return f"# Action: {action_type}"
def _generate_condition_code(self, node: WorkflowNode) -> str:
"""Generate code for condition node"""
condition = node.config.get("condition", "")
return f"""
if {condition}:
# Condition met
pass
else:
# Condition not met
pass
"""
def execute_workflow(self, nodes: List[WorkflowNode]) -> Dict[str, Any]:
"""Execute visual workflow"""
try:
# Translate to code
code = self.translate_workflow(nodes)
# Execute
result = self.sandbox.run_code(code, timeout=60)
return {
"success": result.success,
"stdout": result.stdout,
"stderr": result.stderr,
"generated_code": code
}
except Exception as e:
return {
"success": False,
"error": str(e)
}
def cleanup(self):
"""Clean up backend resources"""
if self.sandbox:
self.sandbox.kill()
self.sandbox = None
# Usage
backend = LowCodeBackend(api_key=os.getenv("HOPX_API_KEY"))
backend.initialize()
# Define workflow
nodes = [
WorkflowNode("node_1", "trigger", {"name": "webhook"}),
WorkflowNode("node_2", "action", {
"action_type": "http_request",
"method": "GET",
"url": "https://api.example.com/data"
})
]
result = backend.execute_workflow(nodes)
print(result)
backend.cleanup()
