Skip to main content

sdk.datalayer

Datalayer AI SDK - A simple SDK for AI engineers to work with Datalayer.\nProvides authentication, runtime creation, and code execution capabilities.

DEFAULT_TIMEOUT​

Minutes

DatalayerClientAuthMixin Objects​

class DatalayerClientAuthMixin()

Mixin class for Datalayer client authentication.\n\nProvides methods to authenticate and fetch resources from the Datalayer server.

DatalayerClient Objects​

class DatalayerClient(DatalayerClientAuthMixin, RuntimesMixin,
                      EnvironmentsMixin, SecretsMixin, SnapshotsMixin)

SDK for Datalayer AI platform.\n\nProvides a unified interface for authentication, runtime creation,\nand code execution in Datalayer environments.

__init__​

def __init__(run_url: str = DEFAULT_RUN_URL, token: Optional[str] = None)

'Initialize the Datalayer SDK.\n\nParameters\n----------\nrun_url:\n Datalayer server URL. Defaults to "https://prod1.datalayer.run".\ntoken:\n Authentication token (can also be set via DATALAYER_TOKEN env var.')

authenticate​

def authenticate() -> bool

Validate authentication credentials.\n\nReturns:\n\n- bool - True if authentication is successful

list_environments​

@lru_cache
def list_environments() -> list["Environment"]

List all available environments.\n\nReturns\n-------\n list: A list of available environments.

create_runtime​

def create_runtime(name: Optional[str] = None,
                   environment: str = DEFAULT_ENVIRONMENT,
                   timeout: float = DEFAULT_TIMEOUT,
                   snapshot_name: Optional[str] = None) -> "Runtime"

'Create a new runtime (kernel for code execution.\n\nParameters\n----------\nname:\n Name of the kernel to create (default: python3)\nenvionment:\n Type of resources needed (cpu, gpu, etc.)\ntimeout:\n Request timeout in minutes. Defaults to 10 minutes.\nsnapshot_name: Optional[str]\n Name of the snapshot to create from. If provided, the runtime will be created from this snapshot.\nReturns\n-------\n Runtime: A runtime object for code execution')

list_runtimes​

def list_runtimes() -> list["Runtime"]

Terminate a running Runtime.\n\nParameters\n----------\nruntime: Runtime or str\n\n'environment_title':'AI Environment',\n'type': 'notebook',

terminate_runtime​

def terminate_runtime(runtime: Union["Runtime", str]) -> bool

Terminate a running Runtime.\n\nParameters\n----------\nruntime: Runtime or str

list_secrets​

def list_secrets() -> list["Secret"]

List all secrets available in the Datalayer environment.\n\nReturns\n-------\n List[Secret]: A list of Secret objects.

create_secret​

def create_secret(name: str,
                  description: str,
                  value: str,
                  secret_type: str = SecretType.GENERIC) -> "Secret"

Create a new secret.\n\nParameters\n----------\nname: str\n Name of the secret.\ndescription: str\n Description of the secret.\nvalue: str\n Value of the secret.\nsecret_type: str\n Type of the secret (e.g., "generic", "password", "key", "token.\n\nReturns\n-------\n Secret: The created secret object.')

delete_secret​

def delete_secret(secret: Union[str, "Secret"]) -> dict[str, str]

Delete a secret by its unique identifier.\n\nParameters\n----------\nsecret: Union[str, Secret]\n Unique identifier of the secret or a Secret object.

create_snapshot​

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

Create a snapshot of the current runtime state.\n\nParameters\n----------\nruntime: Optional[Runtime]\n The runtime object to create a snapshot from.\npod_name: Optional[str]\n The pod name of the runtime.\nname: str, optional\n Name for the new snapshot.\ndescription: str, optional\n Description for the new snapshot.

list_snapshots​

def list_snapshots() -> list["RuntimeSnapshot"]

List all snapshots.\n\nReturns\n-------\nlist: A list of snapshots associated.

delete_snapshot​

def delete_snapshot(snapshot: Union[str, "RuntimeSnapshot"]) -> dict[str, str]

Delete a specific snapshot.\n\nReturns\n-------\ndict: The result of the deletion operation.

Environment Objects​

class Environment()

Represents a Datalayer environment.\n\n [\n {\n "contents": [\n {\n "mount": "/home/jovyan/ai-models",\n "name": "nfs-ai-models-content"\n },\n {\n "mount": "/home/jovyan/data/datalayer-curated",\n "name": "s3-datalayer-datasets",\n "permissions": "ro"\n },\n {\n "mount": "/home/jovyan/data/aws-opendata/satellite-sentinels",\n "name": "s3-satellite-sentinels-images-content",\n "permissions": "ro"\n },\n {\n "mount": "/home/jovyan/data/aws-opendata/genome-browser",\n "name": "s3-genome-browser-content",\n "permissions": "ro"\n }\n ],\n "description": "
\n\n

An environment for LLM inference, AI fine-tuning & training.

\n

GPU detail: 1 NVIDIA A100 80GB

\n

Packages: llama-cpp, pytorch, torchvision, fastai, langchain, huggingface-hub, transformers...

\n
\n",\n "dockerImage": "qnnxrq3e.c1.bhs5.container-registry.ovh.net/datalayer/jupyter-ai-cuda:0.1.0",\n "example": "https://raw.githubusercontent.com/datalayer/examples/5a65d7403de2c34f2052526360eeda84aaceabc6/fastai-classifier/fastai-image-classification.ipynb",\n "kernel": {\n "givenNameTemplate": "AI Runtime"\n },\n "language": "python",\n "owner": "datalayer",\n "resourcesRanges": {\n "default": {\n "limits": {\n "cpu": "10",\n "ephemeral-storage": "20Gi",\n "memory": "80Gi",\n "nvidia.com/gpu": "1"\n },\n "requests": {\n "cpu": "10",\n "ephemeral-storage": "20Gi",\n "memory": "80Gi",\n "nvidia.com/gpu": "1"\n }\n }\n },\n "snippets": [\n {\n "code": "from transformers import AutoTokenizer, AutoModelForCausalLM\nimport torch\nprompt = """\nWhat color is the sky?\n"""\ntokenizer = AutoTokenizer.from_pretrained("mistralai/Mixtral-8x22B-Instruct-v0.1\nmodel = AutoModelForCausalLM.from_pretrained(\n "mistralai/Mixtral-8x22B-Instruct-v0.1",\n device_map="auto",\n torch_dtype=torch.bfloat16,\n)\nchat = [\n &#123;"role": "user", "content": prompt&#125;,\n]\nprompt = tokenizer.apply_chat_template(chat, tokenize=False, add_generation_prompt=True)\ninputs = tokenizer.encode(prompt, add_special_tokens=False, return_tensors="pt")\noutputs = model.generate(input_ids=inputs.to(model.device), max_new_tokens=2000)\nprompt_length = inputs.shape[1]\nresponse = tokenizer.decode(outputs[0][prompt_length:])\nprint(response)\n#\nfrom llama_cpp import Llama\n__Set gpu_layers to the number of layers to offload to GPU. Set to 0 if no acceleration is available on your system.\n\nllm = Llama(\n model_path="./Mixtral-8x7B-Instruct-v0.1.Q4_K_M.gguf", # Download the model file first\n n_ctx=2048, # The max sequence length to use - note that longer sequence lengths require much more resources\n n_threads=8, # The number of CPU threads to use, tailor to your system and the resulting performance\n n_gpu_layers=15, # The number of layers to offload to GPU, if you have acceleration available - try with 35\n verbose = True,\n)\n__Simple inference...\n\nprompt = "who are you?"\noutput = llm(\n "[INST] &#123;prompt&#125; [/INST]", # Prompt\n max_tokens=512, # Generate up to 512 tokens\n stop=["</s>"], # Example stop token - not necessarily correct for this specific model! Please check before using.\n echo=True # Whether to echo the prompt\n)\nprint(output)\n__Chat Completion example...__\n\nllm = Llama(model_path="./Mixtral-8x7B-Instruct-v0.1.Q4_K_M.gguf", chat_format="llama-2") # Set chat_format according to the model you are using.\nllm.create_chat_completion(\n messages = [\n &#123;"role": "system", "content": "You are a story writing assistant."&#125;,\n &#123;\n "role": "user",\n "content": "Write a story about llamas."\n &#125;\n ]\n)\n",\n "title": "ai-prompting"\n &#125;\n ],\n "title": "AI Environment",\n "visibility": "public",\n "name": "ai-env",\n "burning_rate": 0.01,\n "resources": &#123;\n "cpu": "10",\n "ephemeral-storage": "20Gi",\n "memory": "80Gi",\n "nvidia.com/gpu": "1"\n &#125;\n &#125;,\n &#123;\n "contents": [\n &#123;\n "mount": "/home/jovyan/ai-models",\n "name": "nfs-ai-models-content"\n &#125;,\n &#123;\n "mount": "/home/jovyan/data/datalayer-curated",\n "name": "s3-datalayer-datasets",\n "permissions": "ro"\n &#125;,\n &#123;\n "mount": "/home/jovyan/data/aws-opendata/satellite-sentinels",\n "name": "s3-satellite-sentinels-images-content",\n "permissions": "ro"\n &#125;,\n &#123;\n "mount": "/home/jovyan/data/aws-opendata/genome-browser",\n "name": "s3-genome-browser-content",\n "permissions": "ro"\n &#125;\n ],\n "description": "<img src="https://images.datalayer.io/icons/python-100.png" width="60"/>\n<p><b>A Python environment for heavy data processing, parallel analysis</b></p>\n<p><i>Packages: numpy, pandas, matplotlib, seaborn, scikit-learn, dask, ...</i></p>\n<br/>\n",\n "dockerImage": "qnnxrq3e.c1.bhs5.container-registry.ovh.net/datalayer/jupyter-python:0.1.0",\n "example": "https://raw.githubusercontent.com/datalayer/examples/5a65d7403de2c34f2052526360eeda84aaceabc6/opencv-face-detection/youtube-face-detection.ipynb",\n "kernel": &#123;\n "givenNameTemplate": "Python CPU Runtime"\n &#125;,\n "language": "python",\n "owner": "datalayer",\n "resourcesRanges": &#123;\n "default": &#123;\n "limits": &#123;\n "cpu": "4",\n "ephemeral-storage": "20Gi",\n "memory": "20Gi"\n &#125;,\n "requests": &#123;\n "cpu": "2",\n "ephemeral-storage": "20Gi",\n "memory": "10Gi"\n &#125;\n &#125;\n &#125;,\n "title": "Python CPU Environment",\n "visibility": "public",\n "name": "python-cpu-env",\n "burning_rate": 0.008,\n "resources": &#123;\n "cpu": "2",\n "ephemeral-storage": "20Gi",\n "memory": "10Gi"\n &#125;\n &#125;\n ]')

__init__​

def __init__(name: str,
             title: str,
             burning_rate: float,
             language: str,
             owner: str,
             visibility: str,
             metadata: Optional[dict[str, Any]] = None)

Initialize an environment.\n\nParameters\n----------\nname: str\n Name of the environment.\ntitle: str\n Title of the environment.\nburning_rate: float\n The cost of running the environment per hour.

Response Objects​

class Response()

Represents the response from code execution in a runtime.

__init__​

def __init__(execute_response: list[dict[str, Any]])

Initialize a response object.\n\nParameters\n----------\nexecute_response: dict[str, Any]\n The response from the code execution.

stdout​

@property
def stdout() -> str

Get the standard output of the code execution.\n\nReturns\n-------\nstr: The standard output as a string.

stderr​

@property
def stderr() -> str

Get the standard output of the code execution.\n\nReturns\n-------\nstr: The standard output as a string.

Runtime Objects​

class Runtime(DatalayerClientAuthMixin, RuntimesMixin, SnapshotsMixin)

'Represents a Datalayer runtime (kernel for code execution.')

__init__​

def __init__(name: str,
             environment: str = DEFAULT_ENVIRONMENT,
             timeout: float = DEFAULT_TIMEOUT,
             run_url: str = DEFAULT_RUN_URL,
             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[str] = None,
             kernel_token: Optional[str] = None,
             started_at: Optional[str] = None,
             expired_at: Optional[str] = None)

'Initialize a runtime.\n\nParameters\n----------\nname: str\n Name of the runtime (kernel.\nenvironment: str\n Environment type (e.g., "python-cpu-env").\ntimeout: float\n Request timeout in minutes.\nrun_url: str\n Datalayer server URL.\ntoken: Optional[str]\n Authentication token (can also be set via DATALAYER_TOKEN env var).')

__enter__​

def __enter__() -> "Runtime"

Context manager entry.

__exit__​

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

Context manager exit.

get_variable​

def get_variable(name: str) -> Any

Set a variable in the runtime.

set_variable​

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

Set a variable in the runtime.

set_variables​

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

Set variables in the runtime.

execute_file​

def execute_file(path: Union[str, Path],
                 variables: Optional[dict[str, Any]] = None,
                 output: Optional[str] = None) -> Response

Execute a Python file in the runtime.\n\nParameters\n----------\npath: str\n Path to the Python file to execute.\nvariables: Optional[dict[str, Any]]\n Optional variables to set before executing the code.\noutput: Optional[str]\n Optional output variable to return as result.\n\nReturns\n-------\n Response: The result of the code execution.

execute_code​

def execute_code(code: str,
                 variables: Optional[dict[str, Any]] = None,
                 output: Optional[str] = None) -> Union[Response, Any]

Execute code in the runtime.\n\nParameters\n----------\ncode: str\n The Python code to execute.\nvariables: Optional[dict[str, Any]]\n Optional variables to set before executing the code.\noutput: Optional[str]\n Optional output variable to return as result.\n\nReturns\n-------\n Response: 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) -> Union[Response, Any]

'Execute code in the runtime.\n\n Parameters\n ----------\n code_or_path: Union[str, Path]\n The Python code or path to the file to execute.\n variables: Optional[dict[str, Any]]\n Optional variables to set before executing the code.\n output: Optional[str]\n Optional output variable to return as result.\n\n Returns\n -------\n dict:\n The result of the code execution.\n\n\n {\n 'execution_count': 1,\n 'outputs': [\n {\n 'output_type': 'error',\n 'ename': 'SyntaxError',\n 'evalue': 'incomplete input (843041379.py, line 1&#x27;,\n &#x27;traceback&#x27;: [&#x27;\x1b[0;36m Cell \x1b[0;32mIn[1], line 1\x1b[0;36m\x1b[0m\n\x1b[0;31m print("Hello, Datalayer!"\x1b[0m\n\x1b[0m ^\x1b[0m\n\x1b[0;31mSyntaxError\x1b[0m\x1b[0;31m:\x1b[0m incomplete input\n&#x27;]\n &#125;\n ],\n &#x27;status&#x27;: &#x27;error&#x27;\n &#125;\n &#125;')

terminate​

def terminate() -> bool

Terminate the Runtime.

create_snapshot​

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

Create a new snapshot from the current state.\nParameters\n----------\nname: str\n Name for the new snapshot.\ndescription: str\n Description for the new snapshot.\nReturns\n-------\nRuntimeSnapshot: A new snapshot object.

Secret Objects​

class Secret()

Represents a secret in Datalayer.\n\nParameters\n----------\nuid: str\n Unique identifier for the secret.\nname: str\n Name of the secret.\ndescription: str\n Description of the secret.\nsecret_type: str\n Type of the secret (e.g., "generic", "password", "key", "token.')

RuntimeSnapshot Objects​

class RuntimeSnapshot()

Represents a snapshot of a Datalayer runtime state.

__init__​

def __init__(uid: str, name: str, description: str, environment: str,
             metadata: dict[str, Any])

Initialize a runtime snapshot.\n\nParameters\n----------\nuid: str\n Unique identifier for the snapshot.\nname: str\n Name of the snapshot.\ndescription: str\n Description of the snapshot.\nenvironment: str\n Environment associated with the snapshot.\nmetadata: dict\n Metadata related to the snapshot.

enviroment​

@property
def enviroment() -> str

Get the environment of the snapshot.\n\nReturns\n-------\nstr: The environment associated with the snapshot.

uid​

@property
def uid() -> str

Get the unique identifier of the snapshot.\n\nReturns\n-------\nstr: The unique identifier of the snapshot.

metadata​

@property
def metadata() -> dict[str, str]

Get the metadata of the snapshot.\n\nReturns\n-------\ndict: The metadata associated with the snapshot.