Skip to main content

runtime

Runtime services for Datalayer.

Provides runtime management and code execution capabilities in Datalayer environments.

class RuntimeService(name: str, environment: str = 'python-cpu-env', time_reservation: float = 10.0, run_url: str = 'https://prod1.datalayer.run', iam_url: Optional[str] = None, 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, jupyter_token: Optional[str] = None, started_at: Optional[str] = None, expired_at: Optional[str] = None)​

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__(name: str, environment: str = 'python-cpu-env', time_reservation: float = 10.0, run_url: str = 'https://prod1.datalayer.run', iam_url: Optional[str] = None, 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, 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., "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.

  • iam_url : Optional[str]

    Datalayer IAM server URL. If not provided, defaults to run_url.

  • token : Optional[str]

    Authentication token (can also be set via DATALAYER_API_KEY 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.

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

Get the runtime model containing all configuration and state data.

Provides access to all runtime properties including:

  • Configuration: name, environment, run_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​

Get a DatalayerURLs object with the configured URLs.

Returns​

  • DatalayerURLs

    URLs object with run_url and iam_url from the runtime configuration.

pod_name​

Get the pod name.

name​

Get the runtime name.

ingress​

Get the ingress URL.

jupyter_token​

Get the kernel token.

expired_at​

Get the expiration time.

environment​

Get the environment name.

reservation_id​

Get the reservation ID.

uid​

Get the runtime UID.

burning_rate​

Get the burning rate.

started_at​

Get the start time.

__del__() -> None​

Clean up resources when the runtime object is deleted.

__enter__() -> 'RuntimeService'​

Context manager entry.

Returns​

  • RuntimesService

    The runtime instance.

Raises​

  • RuntimeError

    If runtime startup fails.

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

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.models.base.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(code: str, variables: Optional[dict[str, Any]] = None, output: Optional[str] = None, debug: bool = False, timeout: float = 10.0) -> Union[datalayer_core.models.base.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(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.models.base.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() -> 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) -> 'RuntimeSnapshotModel'​

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.