Spaces:
Paused
Paused
File size: 5,116 Bytes
4efde5d |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
from daytona_sdk import AsyncDaytona, DaytonaConfig, CreateSandboxFromSnapshotParams, AsyncSandbox, SessionExecuteRequest, Resources, SandboxState
from dotenv import load_dotenv
from utils.logger import logger
from utils.config import config
from utils.config import Configuration
load_dotenv()
logger.debug("Initializing Daytona sandbox configuration")
daytona_config = DaytonaConfig(
api_key=config.DAYTONA_API_KEY,
api_url=config.DAYTONA_SERVER_URL,
target=config.DAYTONA_TARGET,
)
if daytona_config.api_key:
logger.debug("Daytona API key configured successfully")
else:
logger.warning("No Daytona API key found in environment variables")
if daytona_config.api_url:
logger.debug(f"Daytona API URL set to: {daytona_config.api_url}")
else:
logger.warning("No Daytona API URL found in environment variables")
if daytona_config.target:
logger.debug(f"Daytona target set to: {daytona_config.target}")
else:
logger.warning("No Daytona target found in environment variables")
daytona = AsyncDaytona(daytona_config)
async def get_or_start_sandbox(sandbox_id: str) -> AsyncSandbox:
"""Retrieve a sandbox by ID, check its state, and start it if needed."""
logger.debug(f"Getting or starting sandbox with ID: {sandbox_id}")
try:
sandbox = await daytona.get(sandbox_id)
# Check if sandbox needs to be started
if sandbox.state == SandboxState.ARCHIVED or sandbox.state == SandboxState.STOPPED:
logger.debug(f"Sandbox is in {sandbox.state} state. Starting...")
try:
await daytona.start(sandbox)
# Wait a moment for the sandbox to initialize
# sleep(5)
# Refresh sandbox state after starting
sandbox = await daytona.get(sandbox_id)
# Start supervisord in a session when restarting
await start_supervisord_session(sandbox)
except Exception as e:
logger.error(f"Error starting sandbox: {e}")
raise e
logger.debug(f"Sandbox {sandbox_id} is ready")
return sandbox
except Exception as e:
logger.error(f"Error retrieving or starting sandbox: {str(e)}")
raise e
async def start_supervisord_session(sandbox: AsyncSandbox):
"""Start supervisord in a session."""
session_id = "supervisord-session"
try:
logger.debug(f"Creating session {session_id} for supervisord")
await sandbox.process.create_session(session_id)
# Execute supervisord command
await sandbox.process.execute_session_command(session_id, SessionExecuteRequest(
command="exec /usr/bin/supervisord -n -c /etc/supervisor/conf.d/supervisord.conf",
var_async=True
))
logger.debug(f"Supervisord started in session {session_id}")
except Exception as e:
logger.error(f"Error starting supervisord session: {str(e)}")
raise e
async def create_sandbox(password: str, project_id: str = None) -> AsyncSandbox:
"""Create a new sandbox with all required services configured and running."""
logger.debug("Creating new Daytona sandbox environment")
logger.debug("Configuring sandbox with snapshot and environment variables")
labels = None
if project_id:
logger.debug(f"Using sandbox_id as label: {project_id}")
labels = {'id': project_id}
params = CreateSandboxFromSnapshotParams(
snapshot=Configuration.SANDBOX_SNAPSHOT_NAME,
public=True,
labels=labels,
env_vars={
"CHROME_PERSISTENT_SESSION": "true",
"RESOLUTION": "1048x768x24",
"RESOLUTION_WIDTH": "1048",
"RESOLUTION_HEIGHT": "768",
"VNC_PASSWORD": password,
"ANONYMIZED_TELEMETRY": "false",
"CHROME_PATH": "",
"CHROME_USER_DATA": "",
"CHROME_DEBUGGING_PORT": "9222",
"CHROME_DEBUGGING_HOST": "localhost",
"CHROME_CDP": ""
},
resources=Resources(
cpu=2,
memory=4,
disk=5,
),
auto_stop_interval=15,
auto_archive_interval=30,
)
# Create the sandbox
sandbox = await daytona.create(params)
logger.debug(f"Sandbox created with ID: {sandbox.id}")
# Start supervisord in a session for new sandbox
await start_supervisord_session(sandbox)
logger.debug(f"Sandbox environment successfully initialized")
return sandbox
async def delete_sandbox(sandbox_id: str) -> bool:
"""Delete a sandbox by its ID."""
logger.debug(f"Deleting sandbox with ID: {sandbox_id}")
try:
# Get the sandbox
sandbox = await daytona.get(sandbox_id)
# Delete the sandbox
await daytona.delete(sandbox)
logger.debug(f"Successfully deleted sandbox {sandbox_id}")
return True
except Exception as e:
logger.error(f"Error deleting sandbox {sandbox_id}: {str(e)}")
raise e
|