InterpreterService
class InterpreterService()Synchronous InterpreterService.
InterpreterService.__init__
def __init__(impl: AsyncInterpreterService, bridge: SyncBridge) -> NoneInterpreterService.connect_context
def connect_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContextInterpreterService.create_context
def create_context(*,
language: str,
context_id: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
request_timeout: Optional[float] = None) -> InterpreterContextInterpreterService.delete_context
def delete_context(context: InterpreterContext | InterpreterContextJSON | str) -> NoneInterpreterService.get_context
def get_context(context: str | InterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> InterpreterContextInterpreterService.list_contexts
def list_contexts() -> list[InterpreterContext]InterpreterService.run
def run(code: str,
*,
language: str = 'python',
context: Optional[InterpreterContext] = None,
isolated: bool = False,
timeout: Optional[float] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None) -> InterpreterResultExecute code and return the parsed interpreter result.
AsyncInterpreterService
class AsyncInterpreterService()AsyncInterpreterService.__init__
def __init__(sandbox_id: str,
transport: AsyncTransport,
process: AsyncProcessService) -> NoneAsyncInterpreterService.create_context
async def create_context(*,
language: str,
context_id: Optional[str] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
request_timeout: Optional[float] = None) -> AsyncInterpreterContextAsyncInterpreterService.connect_context
def connect_context(context: str | AsyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> AsyncInterpreterContextAsyncInterpreterService.get_context
def get_context(context: str | AsyncInterpreterContext | InterpreterContextJSON,
*,
language: str | None = None) -> AsyncInterpreterContextAsyncInterpreterService.list_contexts
async def list_contexts() -> list[AsyncInterpreterContext]AsyncInterpreterService.delete_context
async def delete_context(context: AsyncInterpreterContext | InterpreterContextJSON | str) -> NoneAsyncInterpreterService.run
async def run(code: str,
*,
language: str = 'python',
context: Optional[AsyncInterpreterContext] = None,
isolated: bool = False,
timeout: Optional[float] = None,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None) -> InterpreterResultExecute code and return the parsed interpreter result.
A rehydrated kind='code' process handle can still replay logs and
wait/kill, but it does not reconstruct this decomposed result on
another worker. A vanished explicit context surfaces the daemon's typed
SESSION_NOT_FOUND error; callers decide whether to recreate the
context and what to rerun. timeout is the caller wait budget in
seconds; 0 waits unbounded.
InterpreterContext
class InterpreterContext()Serializable interpreter context handle.
InterpreterContext.__init__
def __init__(impl: AsyncInterpreterContext, bridge: SyncBridge) -> NoneInterpreterContext.sandbox_id
@property
def sandbox_id() -> strInterpreterContext.context_id
@property
def context_id() -> strInterpreterContext.language
@property
def language() -> strInterpreterContext.close
def close() -> NoneInterpreterContext.run
def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResultExecute code inside this explicit interpreter context.
AsyncInterpreterContext
@dataclass(frozen=True)
class AsyncInterpreterContext()Serializable interpreter context handle.
Attributes:
sandbox_idstrcontext_idstrlanguagestr
AsyncInterpreterContext.bind
def bind(runner: Callable[[str, Optional[str], Optional[Mapping[str, str]], Optional[float]], Awaitable[InterpreterResult]],
closer: Callable[[], Awaitable[None]]) -> SelfAsyncInterpreterContext.run
async def run(code: str,
*,
cwd: Optional[str] = None,
env: Optional[Mapping[str, str]] = None,
timeout: Optional[float] = None) -> InterpreterResultExecute code inside this explicit interpreter context.
If the named context vanished, the daemon's typed
SESSION_NOT_FOUND error surfaces unchanged so the caller can
recreate the context and decide what code, if any, is safe to rerun.
timeout is the caller wait budget in seconds; 0 waits
unbounded.
AsyncInterpreterContext.close
async def close() -> NoneInterpreterContextJSON
class InterpreterContextJSON(TypedDict)Attributes:
sandbox_idstrcontext_idstrlanguagestr
InterpreterDisplay
@dataclass(frozen=True)
class InterpreterDisplay()One rich interpreter output bundle.
Attributes:
mimestrdatastr
InterpreterProcess
@dataclass(frozen=True)
class InterpreterProcess()Attributes:
process_idstrexit_codeOptional[int]signalOptional[str]reasonOptional[WaitReason]raw_reasonOptional[str] = None
InterpreterLogs
@dataclass(frozen=True)
class InterpreterLogs()Attributes:
stdoutstr = ''stderrstr = ''
InterpreterError
@dataclass(frozen=True)
class InterpreterError()Attributes:
namestrvaluestrtracebackList[str] = field(default_factory=list)
InterpreterOutput
@dataclass(frozen=True)
class InterpreterOutput()Attributes:
dataJsonObjectis_main_resultbooltextOptional[str] = NonehtmlOptional[str] = NonepngOptional[str] = NonesvgOptional[str] = NonejsonOptional[JsonValue] = None
InterpreterResult
@dataclass(frozen=True)
class InterpreterResult()Interpreter run result.
Attributes:
processInterpreterProcessresultsList[InterpreterOutput] = field(default_factory=list)logsInterpreterLogs = field(default_factory=InterpreterLogs)errorOptional[InterpreterError] = None