File size: 4,282 Bytes
a51a15b |
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 |
#!/usr/bin/env python
"""
Script to query and delete sandboxes for a given account ID.
Usage:
python delete_user_sandboxes.py <account_id>
"""
import asyncio
import sys
import os
from typing import List, Dict, Any
from dotenv import load_dotenv
# Load script-specific environment variables
load_dotenv(".env")
from services.supabase import DBConnection
from sandbox.sandbox import daytona
from utils.logger import logger
async def get_user_sandboxes(account_id: str) -> List[Dict[str, Any]]:
"""
Query all projects and their sandboxes associated with a specific account ID.
Args:
account_id: The account ID to query
Returns:
List of projects with sandbox information
"""
db = DBConnection()
client = await db.client
# Print the Supabase URL being used
print(f"Using Supabase URL: {os.getenv('SUPABASE_URL')}")
# Query projects by account_id
result = await client.table('projects').select(
'project_id',
'name',
'sandbox'
).eq('account_id', account_id).execute()
# Print the query result for debugging
print(f"Query result: {result}")
if not result.data:
logger.info(f"No projects found for account ID: {account_id}")
return []
# Filter projects with sandbox information
projects_with_sandboxes = [
project for project in result.data
if project.get('sandbox') and project['sandbox'].get('id')
]
logger.info(f"Found {len(projects_with_sandboxes)} projects with sandboxes for account ID: {account_id}")
return projects_with_sandboxes
async def delete_sandboxes(projects: List[Dict[str, Any]]) -> None:
"""
Delete all sandboxes from the provided list of projects.
Args:
projects: List of projects with sandbox information
"""
if not projects:
logger.info("No sandboxes to delete")
return
for project in projects:
sandbox_id = project['sandbox'].get('id')
project_name = project.get('name', 'Unknown')
project_id = project.get('project_id', 'Unknown')
if not sandbox_id:
continue
try:
logger.info(f"Deleting sandbox {sandbox_id} for project '{project_name}' (ID: {project_id})")
# Get the sandbox and delete it
sandbox = daytona.get_current_sandbox(sandbox_id)
daytona.delete(sandbox)
logger.info(f"Successfully deleted sandbox {sandbox_id}")
except Exception as e:
logger.error(f"Error deleting sandbox {sandbox_id}: {str(e)}")
async def main():
"""Main function to run the script."""
if len(sys.argv) != 2:
print(f"Usage: python {sys.argv[0]} <account_id>")
sys.exit(1)
account_id = sys.argv[1]
logger.info(f"Starting sandbox cleanup for account ID: {account_id}")
# Print environment info
print(f"Environment Mode: {os.getenv('ENV_MODE', 'Not set')}")
print(f"Daytona Server: {os.getenv('DAYTONA_SERVER_URL', 'Not set')}")
try:
# Query projects with sandboxes
projects = await get_user_sandboxes(account_id)
# Print sandbox information
for i, project in enumerate(projects):
sandbox_id = project['sandbox'].get('id', 'N/A')
print(f"{i+1}. Project: {project.get('name', 'Unknown')}")
print(f" Project ID: {project.get('project_id', 'Unknown')}")
print(f" Sandbox ID: {sandbox_id}")
# Confirm deletion
if projects:
confirm = input(f"\nDelete {len(projects)} sandboxes? (y/n): ")
if confirm.lower() == 'y':
await delete_sandboxes(projects)
logger.info("Sandbox cleanup completed")
else:
logger.info("Sandbox deletion cancelled")
else:
logger.info("No sandboxes found for deletion")
except Exception as e:
logger.error(f"Error during sandbox cleanup: {str(e)}")
sys.exit(1)
finally:
# Clean up database connection
await DBConnection.disconnect()
if __name__ == "__main__":
asyncio.run(main())
|