Workflows SDK for Python

$ pip install render

$ pip install --upgrade render

from render import Workflows, Retry

# Basic initialization
app = Workflows()

# Initialization with all options set
app = Workflows(
  default_retry=Retry(
    max_retries=3,
    wait_duration_ms=1000,
    backoff_scaling=1.5
  ),
  default_timeout=300,
  default_plan="flex"
)

from render import Workflows
from math_tasks import app as math_app
from text_tasks import app as text_app

app = Workflows.from_workflows(math_app, text_app)

if __name__ == "__main__":
  app.start()

from render import TaskContext, Workflows 

app = Workflows() 

@app.task 
def calculate_square(ctx: TaskContext, a: int) -> int: 
  return a * a

from render import Retry, TaskContext, Workflows

app = Workflows()

@app.task(
  name="calc_square", # Give the task a custom name (defaults to function name)
  retry=Retry( # Define default retry logic for the task
    max_retries=3, # Retry up to 3 times (i.e., 4 total attempts)
    wait_duration_ms=1000, # Set a base retry delay of 1 second
    backoff_scaling=1.5 # Increase delay by 50% after each retry (exponential backoff)
  ),
  timeout_seconds=300, # Timeout in seconds
  plan="2c-4g" # Compute plan
)
def calculate_square(ctx: TaskContext, a: int) -> int:
  return a * a

import asyncio

@app.task
async def sum_squares(ctx: TaskContext, a: int, b: int) -> int:
  first, second = await asyncio.gather(
    ctx.run(calculate_square, a), 
    ctx.run(calculate_square, b) 
  )
  return first + second

ctx.run(sum_squares, 2, 3)

ctx.run(sum_squares, a=2, b=3)

from render import TaskContext, Workflows

app = Workflows()

@app.task
def calculate_square(ctx: TaskContext, a: int) -> int:
  return a * a

if __name__ == "__main__":
  app.start() 

from render import RenderAsync

# Basic initialization
render = RenderAsync()

# Initialization with API key
render = RenderAsync(
  token="rnd_abc123"
)

# Execute the calculate_square task with an input of 2
started_task_run = await render.workflows.start_task(
  "my-workflow/calculate_square",
  [2]
)

task_run_id = started_task_run.id # ID is available immediately
task_run_status = started_task_run.status # Initial status is available immediately

finished_task_run = await started_task_run # Other properties become available after the task run completes
print(finished_task_run.results) # Prints the task run's result, in this case [4]

# Run the calculate_square task and wait for the result
task_run = await render.workflows.run_task(
  "my-workflow/calculate_square",
  [2]
)
print(task_run.results) # [4]

from render.client.types import ListTaskRunsParams

params = ListTaskRunsParams(
  limit=10, # Return up to 10 runs
  cursor="cfQ74cE2sDI=", # Start from this cursor
  owner_id=["tea-d3jm7ai4d50c73fale60"] # Limit to these workspace IDs
)

task_runs = await render.workflows.list_task_runs(params)
for item in task_runs:
  print(item.task_run.id, item.cursor)

await render.workflows.get_task_run("trn-abc123")

await render.workflows.cancel_task_run("trn-abc123")

# Start multiple tasks
run1 = await render.workflows.start_task("my-workflow/add", [1, 2])
run2 = await render.workflows.start_task("my-workflow/add", [5, 8])

# Stream events until all runs complete
pending = {run1.id, run2.id}
async for event in render.workflows.task_run_events(list(pending)):
  print(f"Run {event.id}: status={event.status}")
  pending.discard(event.id)
  if not pending:
    break

started_task_run = await render.workflows.start_task(
  "my-workflow/calculate_square",
  [2]
)
finished_task_run = await started_task_run 

from render import Render

# Basic initialization
render = Render()

# Initialization with API key
render = Render(
  token="rnd_abc123"
)

started_task_run = render.workflows.start_task(
  "my-workflow/calculate_square",
  [2]
)
# Poll with get_task_run(started_task_run.id) or use run_task() to block until done

task_run = render.workflows.run_task("my-workflow/calculate_square", [2])
print(task_run.results)

from render.client.types import ListTaskRunsParams
params = ListTaskRunsParams(limit=10, cursor="", owner_id=["tea-"])
task_runs = render.workflows.list_task_runs(params)
for item in task_runs:
  print(item.task_run.id, item.cursor)

render.workflows.get_task_run("trn-abc123")

render.workflows.cancel_task_run("trn-abc123")

for event in render.workflows.task_run_events(["trn-abc123"]):
  print(f"Run {event.id}: status={event.status}")

started_task_run = await render.workflows.start_task(
  "my-workflow/calculate_square",
  [2]
)
finished_task_run = await started_task_run 

task_run_details = await render.workflows.run_task("my-workflow/calculate_square", [2])

task_run_details = await render.workflows.get_task_run("trn-abc123")

async for event in render.workflows.task_run_events(["trn-abc123"]):
  task_run_details = event

from render.client.errors import (
    RenderError, # Parent class for other exceptions
    ClientError,
    RateLimitError,
    ServerError,
    TimeoutError,
    TaskRunError
)