Spaces:
Build error
Build error
| import docker | |
| import tarfile | |
| import io | |
| from typing import Optional, Dict, List | |
| from src.domain.interfaces import SandboxService | |
| class DockerSandbox(SandboxService): | |
| def __init__(self, image: str = "python:3.9-slim"): | |
| self.client = docker.from_env() | |
| self.image = image | |
| self.containers: Dict[str, docker.models.containers.Container] = {} | |
| async def start_session(self, session_id: str) -> None: | |
| try: | |
| container = self.client.containers.run( | |
| self.image, | |
| detach=True, | |
| tty=True, | |
| name=f"sandbox_{session_id}", | |
| # Add resource limits or network isolation here if needed | |
| ) | |
| self.containers[session_id] = container | |
| except Exception as e: | |
| print(f"Failed to start container for session {session_id}: {e}") | |
| # In a real app, we might want to raise a custom exception | |
| async def stop_session(self, session_id: str) -> None: | |
| container = self.containers.get(session_id) | |
| if container: | |
| try: | |
| container.stop() | |
| container.remove() | |
| del self.containers[session_id] | |
| except Exception as e: | |
| print(f"Failed to stop container for session {session_id}: {e}") | |
| async def execute_shell(self, session_id: str, command: str) -> str: | |
| container = self.containers.get(session_id) | |
| if not container: | |
| # Try to find existing container by name | |
| try: | |
| container = self.client.containers.get(f"sandbox_{session_id}") | |
| self.containers[session_id] = container | |
| except docker.errors.NotFound: | |
| return "Error: Sandbox not running" | |
| try: | |
| exit_code, output = container.exec_run(command) | |
| return output.decode("utf-8") | |
| except Exception as e: | |
| return f"Error executing command: {e}" | |
| async def read_file(self, session_id: str, path: str) -> str: | |
| container = self.containers.get(session_id) | |
| if not container: | |
| try: | |
| container = self.client.containers.get(f"sandbox_{session_id}") | |
| self.containers[session_id] = container | |
| except docker.errors.NotFound: | |
| return "Error: Sandbox not running" | |
| try: | |
| # get_archive returns a tuple (generator, stat) | |
| bits, stat = container.get_archive(path) | |
| file_obj = io.BytesIO() | |
| for chunk in bits: | |
| file_obj.write(chunk) | |
| file_obj.seek(0) | |
| with tarfile.open(fileobj=file_obj) as tar: | |
| # Assuming single file request | |
| member = tar.next() | |
| f = tar.extractfile(member) | |
| return f.read().decode("utf-8") | |
| except Exception as e: | |
| return f"Error reading file: {e}" | |
| import subprocess | |
| import os | |
| class LocalSandbox(SandboxService): | |
| """Fallback sandbox that runs commands locally when Docker is unavailable""" | |
| async def start_session(self, session_id: str) -> None: | |
| # No-op for local execution | |
| pass | |
| async def stop_session(self, session_id: str) -> None: | |
| # No-op for local execution | |
| pass | |
| async def execute_shell(self, session_id: str, command: str) -> str: | |
| try: | |
| # Run command in a subprocess | |
| result = subprocess.run( | |
| command, | |
| shell=True, | |
| capture_output=True, | |
| text=True, | |
| timeout=30 | |
| ) | |
| output = result.stdout | |
| if result.stderr: | |
| output += f"\nStderr: {result.stderr}" | |
| return output | |
| except Exception as e: | |
| return f"Error executing command: {e}" | |
| async def read_file(self, session_id: str, path: str) -> str: | |
| try: | |
| if os.path.exists(path): | |
| with open(path, 'r', encoding='utf-8') as f: | |
| return f.read() | |
| else: | |
| return f"Error: File not found at {path}" | |
| except Exception as e: | |
| return f"Error reading file: {e}" | |
| # Singleton initialization with fallback | |
| try: | |
| docker_sandbox = DockerSandbox() | |
| # Test connection | |
| docker_sandbox.client.ping() | |
| except Exception as e: | |
| print(f"Docker not available ({e}). Falling back to LocalSandbox.") | |
| docker_sandbox = LocalSandbox() | |