SDK Reference

from tensorlake.applications import application, function

# This application's name will be `hello_world`.
@application()
@function()
def hello_world():
    print("Hello, world!")

# This application's name will be `hola_mundo`.
@application()
@function()
def hola_mundo():
    print("Hola, mundo!")

from tensorlake.applications import application, Retries

@application(
    tags={"language": "python"},
    retries=Retries(max_retries=3),
    region="us-east-1",
)
@function()
def hello_world():
    print("Hello, world!")

from tensorlake.applications import function, application

@application()
@function()
def greet(data: str) -> str:
    return data + " from greet!"

from pydantic import BaseModel
from tensorlake.applications import application, function

class PersonSearchQuery(BaseModel):
    name: str
    age: int

class PersonSearchResult(BaseModel):
    matches: list[dict]

@application()
@function()
def process_data(query: PersonSearchQuery | list[PersonSearchQuery], limit: int | None = None) -> PersonSearchResult:
    if isinstance(query, list):
        return PersonSearchResult(
            matches=[
                {"name": q.name, "age": q.age, "id": i+1} for i, q in enumerate(query) if limit is None or i < limit
            ]
        )
    else:
        return PersonSearchResult(matches=[{"name": query.name, "age": query.age, "id": 1}])

class File:
    content: bytes  # Raw bytes of the file
    content_type: str  # MIME content type of the file

from tensorlake.applications import function, application, File

@application()
@function()
def process_file(input: File) -> File:
    print(
        "Got file content type:",
        input.content_type,
        "size:",
         len(input.content),
         "bytes"
    )
    return File(
        # HTTP response body is the raw bytes of the input file.
        content=input.content,
        # HTTP response content type is the same as input file content type.
        content_type=input.content_type
    )

from tensorlake.applications import application, function

# Define an application function which is an HTTP entry point for the application.
@application()
@function()
def greet(name: str) -> str:
    if name.startswith("A"):
        return "Hello, A-name!"
    else:
        # Call another function to perform a specific processing of non-A names.
        return process_non_a_name(name) + " from greet!"

@function()
def process_non_a_name(name: str) -> str:
    return "A" + name[1:]

from tensorlake.applications import function, Image, Retries

@function(
    # Use Ubuntu as a base image instead of the default Debian
    image=Image(base_image="ubuntu:latest"),
    # Make my_secret available to the function as an environment variable
    secrets=["my_secret"],
    # Description of the function in the workflow
    description="Measures the string using its length",
    # Retry the function twice if it fails
    retries=Retries(max_retries=2),
    # Function fails if it was running for more than 30 seconds and didn't report any progress
    timeout=30,
    # 2 CPUs are available to the function
    cpu=2,
    # 4 GB of memory is available to the function
    memory=4,
    # 2 GB of ephemeral /tmp disk space is available to the function
    ephemeral_disk=2,
    # Run the function in a container with GPU support
    gpu="H100",
    # Run the function in us-east-1 region only
    region="us-east-1",
)
def string_length(s: str) -> int:
    return len(s)

from large_model import load_large_model
from tensorlake.applications import application, cls, function, run_remote_application

@cls()
class MyCompute:
    def __init__(self):
        # Run initialization code once on function container startup
        self.model = load_large_model()

    @application()
    @function(cpu=4, memory=16)
    def run(self, data: str) -> int:
        return self.model.run(data)

if __name__ == "__main__":
    run_remote_application("MyCompute.run", data="some input data")

from tensorlake.applications import function

# Set a 30 minute timeout for long-running agent tasks
@function(timeout=1800)
def deep_research(prompt: str) -> str:
    ...

from tensorlake.applications import function, Retries

# Retry the function once if it failed
@function(retries=Retries(max_retries=1))
def my_function() -> int:
    raise Exception("Something went wrong")

from tensorlake.applications import RequestContext, function

@function()
def my_function(data: str) -> int:
    ctx: RequestContext = RequestContext.get()
    print(f"Request ID: {ctx.request_id}")
    ...

from tensorlake.applications import RequestContext, function

@function()
def my_function(data: str) -> str:
    ctx = RequestContext.get()
    print(f"Processing request: {ctx.request_id}")
    return ctx.request_id

from tensorlake.applications import RequestContext, function

@function()
def first_function(data: str) -> int:
    ctx = RequestContext.get()
    ctx.state.set("user_data", data)
    ctx.state.set("processed", True)
    return second_function()

@function()
def second_function() -> int:
    ctx = RequestContext.get()
    data = ctx.state.get("user_data")
    return len(data)

from tensorlake.applications import RequestContext, function

@function()
def process_items(items: list) -> dict:
    ctx = RequestContext.get()
    results = []

    for i, item in enumerate(items):
        ctx.progress.update(i + 1, len(items), f"Processing item {i + 1}")
        results.append(process(item))

    return {"results": results}

import time
from tensorlake.applications import RequestContext, function

@function()
def my_function(data: str) -> int:
    ctx = RequestContext.get()
    start_time = time.monotonic()
    result = len(data)
    
    # Record metrics
    ctx.metrics.timer("processing_time", time.monotonic() - start_time)
    ctx.metrics.counter("items_processed", result)
    
    return result

from tensorlake.applications import function

# Allocate 4 CPUs for data processing
@function(cpu=4)
def process_data(data: list) -> dict:
    ...

from tensorlake.applications import function

# Allocate 8 GB memory for loading large models
@function(memory=8)
def run_model(data: str) -> str:
    ...

from tensorlake.applications import function

# Allocate 20 GB disk for downloading and processing large files
@function(ephemeral_disk=20)
def process_video(url: str) -> str:
    # Download video to /tmp
    # Process and return results
    ...