Key Concepts

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",
    input_serializer="pickle",
    output_serializer="pickle"
)
@function()
def hello_world(input: str):
    print(f"Hello, world! {input}")

from tensorlake.applications import application, function

@application()
@function()
def my_function(data: str) -> int:
    return len(data)

from tensorlake.applications import application, function

@application()
@function()
def my_function(data: str) -> int:
    returned_value: int = other_function(data)
    # returned_value == len(data)
    return returned_value

@function
def other_function(data: str) -> int:
    return len(data)

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 function

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

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


@function()
def process_data(query: PersonSearchQuery) -> PersonSearchResult:
    return PersonSearchResult(matches=[{"name": query.name, "age": query.age, "id": 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 cls, function

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

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

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")

import time
from tensorlake.applications import RequestContext, function

@function()
def initial_function(data: str) -> int:
    ctx: RequestContext = RequestContext.get()
    start_time = time.time()

    # Set request progress to 1 out of 2.
    # This extends the function timeout by resetting it.
    ctx.progress.update(1, 2)
    # Print the request information.
    print(f"Request ID: {ctx.request_id}")
    # Set a request key-value pair. It's availble to the functions in the same request.
    ctx.state.set("my_function_data", data)

    # Set the time that has passed since the function call started.
    ctx.metrics.timer("initial_function", time.time() - start_time)

    # Call the next function and return its result.
    final_result: int = final_function()
    # Mark the function call as almost done by setting its progress.
    ctx.progress.update(2, 2)
    # Return the result.
    # final_result == len(data)
    return final_result

@function()
def final_function() -> int:
    ctx: RequestContext = RequestContext.get()
    # Fetch the state data and calculate its length.
    return len(ctx.request_state.get("my_function_data"))