ChefAdorous's picture
Deploy Code Execution Sandbox with FastAPI and Docker
a89f25d
from pydantic import BaseModel, Field, field_validator
from typing import Optional, Literal, Dict, Any
from enum import Enum
from datetime import datetime
import uuid
class Language(str, Enum):
"""Supported programming languages"""
PYTHON = "python"
JAVASCRIPT = "javascript"
BASH = "bash"
class ExecutionRequest(BaseModel):
"""Request model for code execution"""
code: str = Field(..., description="Code to execute", min_length=1, max_length=50000)
language: Language = Field(..., description="Programming language")
stdin: str = Field(default="", description="Standard input for the code")
timeout: int = Field(default=10, ge=1, le=30, description="Execution timeout in seconds")
memory_limit: int = Field(default=256, ge=64, le=512, description="Memory limit in MB")
@field_validator('code')
@classmethod
def validate_code(cls, v: str) -> str:
if not v.strip():
raise ValueError("Code cannot be empty or whitespace only")
return v
class ExecutionResponse(BaseModel):
"""Response model for code execution"""
stdout: str = Field(default="", description="Standard output")
stderr: str = Field(default="", description="Standard error")
exit_code: int = Field(..., description="Exit code of the process")
execution_time: float = Field(..., description="Execution time in seconds")
error: Optional[str] = Field(default=None, description="Error message if execution failed")
class SandboxConfig(BaseModel):
"""Configuration for sandbox execution"""
max_execution_time: int = Field(default=30, description="Maximum execution time in seconds")
max_memory_mb: int = Field(default=512, description="Maximum memory in MB")
enable_network: bool = Field(default=False, description="Enable network access in sandbox")
read_only_root: bool = Field(default=True, description="Make root filesystem read-only")
max_output_size: int = Field(default=1048576, description="Maximum output size in bytes (1MB)")
class LanguageInfo(BaseModel):
"""Information about a supported language"""
name: str
version: str
image: str
extensions: list[str]
class SessionStatus(str, Enum):
"""Session lifecycle status"""
CREATING = "creating"
READY = "ready"
BUSY = "busy"
STOPPING = "stopping"
ERROR = "error"
class CreateSessionRequest(BaseModel):
"""Request to create a new persistent session"""
metadata: Dict[str, Any] = Field(default_factory=dict, description="User-defined metadata")
timeout_minutes: int = Field(default=30, ge=5, le=120, description="Session idle timeout in minutes")
class SessionResponse(BaseModel):
"""Response with session information"""
session_id: str
container_id: str
volume_name: str
status: SessionStatus
created_at: datetime
last_activity: datetime
timeout_minutes: int
metadata: Dict[str, Any]
files_count: Optional[int] = None
class FileInfo(BaseModel):
"""Information about a file in session"""
filename: str
path: str
size: int
modified_at: datetime
mime_type: str
class FileUploadResponse(BaseModel):
"""Response after file upload"""
filename: str
path: str
size: int
message: str = "File uploaded successfully"
class ExecuteInSessionRequest(BaseModel):
"""Request to execute code in an existing session"""
code: str = Field(..., min_length=1, max_length=50000)
language: Language
working_dir: str = Field(default="/workspace", description="Working directory for execution")
timeout: int = Field(default=30, ge=1, le=120)
class ExecuteFileRequest(BaseModel):
"""Request to execute an uploaded file"""
filepath: str = Field(..., description="Path to file in session workspace")
language: Language
args: list[str] = Field(default_factory=list, description="Command-line arguments")
timeout: int = Field(default=30, ge=1, le=120)