Custom Serverless Platform

from hopx_ai import Sandbox
import os
import json
from typing import Dict, Any, Callable

class ServerlessFunction:
    def __init__(self, function_id: str, code: str, runtime: str = "python"):
        self.function_id = function_id
        self.code = code
        self.runtime = runtime
        self.sandbox = None
    
    def initialize(self, api_key: str):
        """Initialize function sandbox"""
        self.sandbox = Sandbox.create(
            template="code-interpreter",
            api_key=api_key,
            timeout_seconds=300  # 5 minute timeout
        )
    
    def invoke(self, event: Dict[str, Any], context: Dict[str, Any] = None) -> Dict[str, Any]:
        """Invoke function with event"""
        try:
            # Prepare function code with event
            function_code = f"""
import json

# Event data
event = {json.dumps(event)}

# Context data
context = {json.dumps(context or {})}

# User function
{self.code}
"""
            
            # Execute function
            result = self.sandbox.run_code(function_code, timeout=30)
            
            return {
                "success": result.success,
                "result": result.stdout,
                "error": result.stderr if not result.success else None,
                "execution_time": result.execution_time
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e)
            }
    
    def cleanup(self):
        """Clean up function resources"""
        if self.sandbox:
            self.sandbox.kill()
            self.sandbox = None

class ServerlessPlatform:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.functions = {}
    
    def register_function(self, function_id: str, code: str, runtime: str = "python"):
        """Register a serverless function"""
        func = ServerlessFunction(function_id, code, runtime)
        func.initialize(self.api_key)
        self.functions[function_id] = func
    
    def invoke_function(self, function_id: str, event: Dict[str, Any]) -> Dict[str, Any]:
        """Invoke registered function"""
        if function_id not in self.functions:
            return {
                "success": False,
                "error": f"Function {function_id} not found"
            }
        
        return self.functions[function_id].invoke(event)
    
    def cleanup_all(self):
        """Clean up all functions"""
        for func in self.functions.values():
            func.cleanup()

# Usage
platform = ServerlessPlatform(api_key=os.getenv("HOPX_API_KEY"))

# Register function
platform.register_function(
    "hello_world",
    """
def handler(event, context):
    name = event.get('name', 'World')
    return f"Hello, {name}!"
""",
    "python"
)

# Invoke function
result = platform.invoke_function("hello_world", {"name": "HopX"})
print(result)

platform.cleanup_all()

import { Sandbox } from '@hopx-ai/sdk';

class ServerlessFunction {
    constructor(functionId, code, runtime = 'javascript') {
        this.functionId = functionId;
        this.code = code;
        this.runtime = runtime;
        this.sandbox = null;
    }
    
    async initialize(apiKey) {
        this.sandbox = await Sandbox.create({
            template: 'code-interpreter',
            apiKey,
            timeoutSeconds: 300  // 5 minute timeout
        });
    }
    
    async invoke(event, context = null) {
        try {
            // Prepare function code with event
            const functionCode = `
// Event data
const event = ${JSON.stringify(event)};

// Context data
const context = ${JSON.stringify(context || {})};

// User function
${this.code}
`;
            
            // Execute function
            const result = await this.sandbox.runCode(functionCode, {
                language: this.runtime,
                timeout: 30
            });
            
            return {
                success: result.success,
                result: result.stdout,
                error: !result.success ? result.stderr : null,
                executionTime: result.execution_time
            };
        } catch (error) {
            return {
                success: false,
                error: error.message
            };
        }
    }
    
    async cleanup() {
        if (this.sandbox) {
            await this.sandbox.kill();
            this.sandbox = null;
        }
    }
}

class ServerlessPlatform {
    constructor(apiKey) {
        this.apiKey = apiKey;
        this.functions = {};
    }
    
    async registerFunction(functionId, code, runtime = 'javascript') {
        const func = new ServerlessFunction(functionId, code, runtime);
        await func.initialize(this.apiKey);
        this.functions[functionId] = func;
    }
    
    async invokeFunction(functionId, event) {
        if (!this.functions[functionId]) {
            return {
                success: false,
                error: `Function ${functionId} not found`
            };
        }
        
        return await this.functions[functionId].invoke(event);
    }
    
    async cleanupAll() {
        await Promise.all(
            Object.values(this.functions).map(func => func.cleanup())
        );
    }
}

// Usage
const platform = new ServerlessPlatform(process.env.HOPX_API_KEY);

// Register function
await platform.registerFunction(
    'hello_world',
    `
function handler(event, context) {
    const name = event.name || 'World';
    return \`Hello, \${name}!\`;
}
`,
    'javascript'
);

// Invoke function
const result = await platform.invokeFunction('hello_world', { name: 'HopX' });
console.log(result);

await platform.cleanupAll();

class OptimizedServerlessPlatform(ServerlessPlatform):
    def __init__(self, api_key: str):
        super().__init__(api_key)
        self.warm_pool = {}  # Keep sandboxes warm
    
    def _warm_function(self, function_id: str):
        """Warm up function sandbox"""
        if function_id not in self.warm_pool:
            func = self.functions[function_id]
            # Keep sandbox alive
            self.warm_pool[function_id] = func.sandbox
    
    def invoke_function_optimized(self, function_id: str, event: Dict[str, Any]) -> Dict[str, Any]:
        """Invoke with cold start optimization"""
        # Warm function if not already warm
        if function_id not in self.warm_pool:
            self._warm_function(function_id)
        
        return self.invoke_function(function_id, event)

# Usage
platform = OptimizedServerlessPlatform(api_key=os.getenv("HOPX_API_KEY"))
platform.register_function("my_function", "def handler(e, c): return 'OK'")
platform._warm_function("my_function")  # Pre-warm

result = platform.invoke_function_optimized("my_function", {})
print(result)

class OptimizedServerlessPlatform extends ServerlessPlatform {
    constructor(apiKey) {
        super(apiKey);
        this.warmPool = {};  // Keep sandboxes warm
    }
    
    async warmFunction(functionId) {
        if (!this.warmPool[functionId]) {
            const func = this.functions[functionId];
            // Keep sandbox alive
            this.warmPool[functionId] = func.sandbox;
        }
    }
    
    async invokeFunctionOptimized(functionId, event) {
        // Warm function if not already warm
        if (!this.warmPool[functionId]) {
            await this.warmFunction(functionId);
        }
        
        return await this.invokeFunction(functionId, event);
    }
}

// Usage
const platform = new OptimizedServerlessPlatform(process.env.HOPX_API_KEY);
await platform.registerFunction('my_function', "function handler(e, c) { return 'OK'; }");
await platform.warmFunction('my_function');  // Pre-warm

const result = await platform.invokeFunctionOptimized('my_function', {});
console.log(result);