Multi-Tenant Code Execution Platform

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

class Tenant:
    def __init__(self, tenant_id: str, quota: Dict[str, int]):
        self.tenant_id = tenant_id
        self.quota = quota  # {"executions_per_hour": 100, "max_execution_time": 60}

class MultiTenantPlatform:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.tenants = {}
        self.tenant_sandboxes = {}
        self.usage_tracking = {}
    
    def register_tenant(self, tenant: Tenant):
        """Register a tenant"""
        self.tenants[tenant.tenant_id] = tenant
        self.usage_tracking[tenant.tenant_id] = {
            "executions": 0,
            "execution_time": 0
        }
    
    def execute_for_tenant(self, tenant_id: str, code: str) -> Dict[str, Any]:
        """Execute code for specific tenant"""
        if tenant_id not in self.tenants:
            return {
                "success": False,
                "error": f"Tenant {tenant_id} not found"
            }
        
        tenant = self.tenants[tenant_id]
        usage = self.usage_tracking[tenant_id]
        
        # Check quota
        if usage["executions"] >= tenant.quota.get("executions_per_hour", 100):
            return {
                "success": False,
                "error": "Execution quota exceeded"
            }
        
        sandbox = None
        try:
            # Create or get tenant sandbox (isolated per tenant)
            if tenant_id not in self.tenant_sandboxes:
                self.tenant_sandboxes[tenant_id] = Sandbox.create(
                    template="code-interpreter",
                    api_key=self.api_key,
                    timeout_seconds=tenant.quota.get("max_execution_time", 60)
                )
            
            sandbox = self.tenant_sandboxes[tenant_id]
            
            # Set tenant context
            sandbox.env.set("TENANT_ID", tenant_id)
            
            # Execute code
            result = sandbox.run_code(code, timeout=tenant.quota.get("max_execution_time", 60))
            
            # Update usage
            usage["executions"] += 1
            usage["execution_time"] += result.execution_time
            
            return {
                "success": result.success,
                "stdout": result.stdout,
                "stderr": result.stderr,
                "tenant_id": tenant_id,
                "usage": usage
            }
        except Exception as e:
            return {
                "success": False,
                "error": str(e),
                "tenant_id": tenant_id
            }
    
    def get_tenant_usage(self, tenant_id: str) -> Dict[str, Any]:
        """Get tenant usage statistics"""
        if tenant_id not in self.usage_tracking:
            return {"error": "Tenant not found"}
        
        return self.usage_tracking[tenant_id]
    
    def cleanup_tenant(self, tenant_id: str):
        """Clean up tenant resources"""
        if tenant_id in self.tenant_sandboxes:
            self.tenant_sandboxes[tenant_id].kill()
            del self.tenant_sandboxes[tenant_id]

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

# Register tenants
platform.register_tenant(Tenant("tenant_1", {"executions_per_hour": 100, "max_execution_time": 60}))
platform.register_tenant(Tenant("tenant_2", {"executions_per_hour": 200, "max_execution_time": 120}))

# Execute for tenant
result = platform.execute_for_tenant("tenant_1", "print('Hello from tenant 1')")
print(result)

usage = platform.get_tenant_usage("tenant_1")
print(usage)