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=["example", "python"],
    retries=Retries(max_attempts=3),
    region="EU",
    input_serializer="pickle",
    output_serializer="pickle"
)
@function()
def hello_world():
    print("Hello, world!")

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:
    return other_function(data)

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

from tensorlake.applications import function

@function()
def greet(data: str) -> str:
    return data

from pydantic import BaseModel
from tensorlake.applications import function

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

@function()
def process_data(data: MyData) -> str:
    return data.name + " is " + str(data.age) + " years old"

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,
    # Reuse previous function outputs if the function gets called with the same inputs
    cacheable=True,
    # 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="nvidia-tesla-k80",
    # Run the function in the EU region
    region="EU",
)
def string_length(s: str) -> int:
    return len(s)

from large_model import load_large_model
from tensorlake.applications import cls

@cls()
class MyCompute:
    # The same attributes as for @tensorlake_function() decorator
    image = ...
    input_encoder = ...

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

    @function()
    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.
    ctx.progress.update(1, 2)
    # Print the request information.
    print(f"Request ID: {ctx.request_id}, Graph name: {ctx.graph_name}, Graph version: {ctx.graph_version}")
    # Set a request key-value pair. It's availble to the function 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 to complete the request.
    return final_function()

@function()
def final_function() -> int:
    ctx: RequestContext = RequestContext.get()
    # Set the request progress to 2 out of 2.
    ctx.progress.update(2, 2)

    # Fetch the state data and calculate its length.
    return len(ctx.request_state.get("my_function_data"))