Skip to main content

runtimes.runtime_service

Runtime services for Datalayer.

Provides runtime management and code execution capabilities in Datalayer environments.

RuntimeService Objects​

class RuntimeService(AuthnMixin, RuntimesMixin, SandboxSnapshotsMixin)

Service for managing Datalayer runtime operations.

This service handles runtime lifecycle operations such as starting, stopping, code execution, and variable management. The runtime data is managed through the RuntimeModel.

Parameters​

runtime_model : RuntimeModel The runtime model containing all configuration and state data.

__init__​

def __init__(name: str,
             environment: str = DEFAULT_ENVIRONMENT,
             time_reservation: Minutes = DEFAULT_TIME_RESERVATION,
             datalayer_url: str = DEFAULT_DATALAYER_URL,
             iam_url: Optional[str] = None,
             token: Optional[str] = None,
             api_key: Optional[str] = None,
             pod_name: Optional[str] = None,
             ingress: Optional[str] = None,
             reservation_id: Optional[str] = None,
             uid: Optional[str] = None,
             burning_rate: Optional[CreditsPerSecond] = None,
             jupyter_token: Optional[str] = None,
             started_at: Optional[str] = None,
             expired_at: Optional[str] = None)

Initialize a runtime service.

Parameters​

name : str Name of the runtime (kernel). environment : str Environment type (e.g., "ai-agents-env"). Type of resources needed (cpu, gpu, etc.). time_reservation : Minutes Time reservation in minutes for the runtime. Defaults to 10 minutes. datalayer_url : str Datalayer server URL. iam_url : Optional[str] Datalayer IAM server URL. If not provided, defaults to datalayer_url. token : Optional[str] Authentication token (can also be set via DATALAYER_API_KEY env var). api_key : Optional[str] Authentication API key alias for token. pod_name : Optional[str] Name of the pod running the runtime. ingress : Optional[str] Ingress URL for the runtime. reservation_id : Optional[str] Reservation ID for the runtime. uid : Optional[str] ID for the runtime. burning_rate : Optional[float] Burning rate for the runtime. jupyter_token : Optional[str] Token for the kernel client. started_at : Optional[str] Start time for the runtime. expired_at : Optional[str] Expiration time for the runtime.

model​

@property
def model() -> RuntimeModel

Get the runtime model containing all configuration and state data.

Provides access to all runtime properties including:

  • Configuration: name, environment, datalayer_url, iam_url
  • Authentication: token, external_token
  • Runtime state: kernel_client, kernel_id, executing
  • Infrastructure: pod_name, ingress, uid, reservation_id

Returns​

RuntimeModel The runtime model with all runtime data and configuration.

urls​

@property
def urls() -> DatalayerURLs

Get a DatalayerURLs object with the configured URLs.

Returns​

DatalayerURLs URLs object with datalayer_url and iam_url from the runtime configuration.

pod_name​

@property
def pod_name() -> Optional[str]

Get the pod name.

name​

@property
def name() -> str

Get the runtime name.

ingress​

@property
def ingress() -> Optional[str]

Get the ingress URL.

jupyter_token​

@property
def jupyter_token() -> Optional[str]

Get the kernel token.

expired_at​

@property
def expired_at() -> Optional[str]

Get the expiration time.

environment​

@property
def environment() -> str

Get the environment name.

reservation_id​

@property
def reservation_id() -> Optional[str]

Get the reservation ID.

uid​

@property
def uid() -> Optional[str]

Get the runtime UID.

burning_rate​

@property
def burning_rate() -> Optional[float]

Get the burning rate.

started_at​

@property
def started_at() -> Optional[str]

Get the start time.

__del__​

def __del__() -> None

Clean up resources when the runtime object is deleted.

__enter__​

def __enter__() -> "RuntimeService"

Context manager entry.

Returns​

RuntimesService The runtime instance.

Raises​

RuntimeError If runtime startup fails.

__exit__​

def __exit__(exc_type: Any, exc_val: Any, exc_tb: Any) -> None

Context manager exit.

Parameters​

exc_type : Any Exception type. exc_val : Any Exception value. exc_tb : Any Exception traceback.

start​

def start() -> None

Start the runtime and kernel client.

This is a public wrapper for _start() to support non-context usage.

stop​

def stop() -> bool

Stop the runtime and terminate the kernel client.

This is a public wrapper for _stop() to support non-context usage.

get_variable​

def get_variable(name: str) -> Any

Get a variable from the runtime.

Parameters​

name : str Name of the variable to retrieve.

Returns​

Any Value of the variable, or None if not found or runtime not started.

set_variable​

def set_variable(name: str, value: Any) -> None

Set a variable in the runtime.

Parameters​

name : str Name of the variable to set. value : Any Value to assign to the variable.

Returns​

Response Response object containing execution results.

set_variables​

def set_variables(variables: dict[str, Any]) -> None

Set variables in the runtime.

Parameters​

variables : dict[str, Any] Dictionary of variable names and values to set.

Returns​

Response Response object containing execution results.

execute_file​

def execute_file(path: Union[str, Path],
                 variables: Optional[dict[str, Any]] = None,
                 output: Optional[str] = None,
                 debug: bool = False,
                 timeout: Seconds = 10.0) -> ExecutionResponse

Execute a Python file in the runtime.

Parameters​

path : Union[str, Path] Path to the Python file to execute. variables : Optional[dict[str, Any]] Optional variables to set before executing the code. output : Optional[str] Optional output variable to return as result. debug : bool Whether to enable debug mode. If True, the output and error streams will be printed. timeout : Seconds Timeout for the execution.

Returns​

Response The result of the code execution.

execute_code​

def execute_code(code: str,
                 variables: Optional[dict[str, Any]] = None,
                 output: Optional[str] = None,
                 debug: bool = False,
                 timeout: Seconds = 10.0) -> Union[ExecutionResponse, Any]

Execute code in the runtime.

Parameters​

code : str The Python code to execute. variables : Optional[dict[str, Any]] Optional variables to set before executing the code. output : Optional[str] Optional output variable to return as result. debug : bool Whether to enable debug mode. If True, the output and error streams will be printed. timeout : Seconds Timeout for the execution.

Returns​

Union[Response, Any] The result of the code execution.

execute​

def execute(code_or_path: Union[str, Path],
            variables: Optional[dict[str, Any]] = None,
            output: Optional[str] = None,
            debug: bool = False,
            timeout: Seconds = 10.0) -> Union[ExecutionResponse, Any]

Execute code in the runtime.

Parameters​

code_or_path : Union[str, Path] The Python code or path to the file to execute. variables : Optional[dict[str, Any]] Optional variables to set before executing the code. output : Optional[str] Optional output variable to return as result. debug : bool Whether to enable debug mode. If True, the output and error streams will be printed. timeout : Seconds Timeout for the execution.

Returns​

Union[Response, Any] The result of the code execution.

terminate​

def terminate() -> bool

Terminate the Runtime.

Returns​

bool True if termination was successful, False otherwise.

create_snapshot​

def create_snapshot(name: Optional[str] = None,
                    description: Optional[str] = None,
                    stop: bool = True) -> "SandboxSnapshotModel"

Create a new snapshot from the current state.

Parameters​

name : Optional[str] Name for the new snapshot. description : Optional[str] Description for the new snapshot. stop : bool Whether to stop the runtime after creating the snapshot.

Returns​

SandboxSnapshot A new snapshot object.