Skip to main content

runtimes

Runtime classes for the Datalayer SDK.

Provides runtime management and code execution capabilities in Datalayer environments.

class Runtime(name: str, environment: str = 'python-cpu-env', time_reservation: float = 10.0, run_url: str = 'https://prod1.datalayer.run', token: Optional[str] = None, pod_name: Optional[str] = None, ingress: Optional[str] = None, reservation_id: Optional[str] = None, uid: Optional[str] = None, burning_rate: Optional[float] = None, kernel_token: Optional[str] = None, started_at: Optional[str] = None, expired_at: Optional[str] = None)​

Represents a Datalayer runtime (kernel) for code execution.

Parameters​

  • name : str

    Name of the runtime (kernel).

  • environment : str

    Environment type (e.g., "python-cpu-env").

  • time_reservation : Minutes

    Time reservation in minutes for the runtime.

  • run_url : str

    Datalayer server URL.

  • token : Optional[str]

    Authentication token (can also be set via DATALAYER_TOKEN env var).

  • pod_name : Optional[str]

    Pod name for existing runtime.

  • ingress : Optional[str]

    Ingress URL for the runtime.

  • reservation_id : Optional[str]

    Reservation ID for the runtime.

  • uid : Optional[str]

    Unique identifier for the runtime.

  • burning_rate : Optional[str]

    Cost rate for the runtime.

  • kernel_token : Optional[str]

    Kernel authentication token.

  • started_at : Optional[str]

    Runtime start timestamp.

  • expired_at : Optional[str]

    Runtime expiration timestamp.

__init__(name: str, environment: str = 'python-cpu-env', time_reservation: float = 10.0, run_url: str = 'https://prod1.datalayer.run', token: Optional[str] = None, pod_name: Optional[str] = None, ingress: Optional[str] = None, reservation_id: Optional[str] = None, uid: Optional[str] = None, burning_rate: Optional[float] = None, kernel_token: Optional[str] = None, started_at: Optional[str] = None, expired_at: Optional[str] = None)​

Initialize a runtime.

Parameters​

  • name : str

    Name of the runtime (kernel).

  • environment : str

    Environment type (e.g., "python-cpu-env"). Type of resources needed (cpu, gpu, etc.).

  • time_reservation : Minutes

    Time reservation in minutes for the runtime. Defaults to 10 minutes.

  • run_url : str

    Datalayer server URL.

  • token : Optional[str]

    Authentication token (can also be set via DATALAYER_TOKEN env var).

  • 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.

  • kernel_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.

__del__() -> None​

Clean up resources when the runtime object is deleted.

__enter__() -> 'Runtime'​

Context manager entry.

Returns​

  • Runtime

    The runtime instance.

__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.

run_url​

Get the runtime server URL.

Returns​

  • str

    The Datalayer server URL for this runtime.

name​

Get the runtime name.

Returns​

  • str

    The name of this runtime.

uid​

Get the runtime unique identifier.

Returns​

  • Optional[str]

    The unique identifier of this runtime, or None if not set.

pod_name​

Get the runtime pod name.

Returns​

  • Optional[str]

    The pod name of this runtime, or None if not set.

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(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(variables: dict[str, typing.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(path: Union[str, pathlib.Path], variables: Optional[dict[str, Any]] = None, output: Optional[str] = None, debug: bool = False, timeout: float = 10.0) -> datalayer_core.sdk.response.Response​

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(code: str, variables: Optional[dict[str, Any]] = None, output: Optional[str] = None, debug: bool = False, timeout: float = 10.0) -> Union[datalayer_core.sdk.response.Response, 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(code_or_path: Union[str, pathlib.Path], variables: Optional[dict[str, Any]] = None, output: Optional[str] = None, debug: bool = False, timeout: float = 10.0) -> Union[datalayer_core.sdk.response.Response, 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() -> bool​

Terminate the Runtime.

Returns​

  • bool

    True if termination was successful, False otherwise.

create_snapshot(name: Optional[str] = None, description: Optional[str] = None, stop: bool = True) -> 'RuntimeSnapshot'​

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​

  • RuntimeSnapshot

    A new snapshot object.