Defining Workflow Tasks

Specify units of work to run on Render.

Render Workflows is in limited early access.

During the early access period, the Workflows API and SDKs might introduce breaking changes.

After you create your first workflow, you can start defining your own tasks. This article describes supported syntax and configuration options.

First: Install the Render SDK

The Render SDK is currently available only for Python.

TypeScript support is nearing release (see GitHub).

SDKs for additional languages are planned for future releases.

The Render SDK is required to define and register workflow tasks.

From your Python project directory:

shell

$ pip install render_sdk

Make sure to add render_sdk as a dependency to your application's requirements.txt, pyproject.toml, or equivalent.

Basic example

Let's start with a "minimum viable workflow" that defines a single task:

main.py

python

from render_sdk import Workflows

app = Workflows()

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

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

This includes everything required to define a workflow:

We import the Workflows class from the Render SDK and initialize it as app.
We apply the @app.task decorator to a function (calculate_square) to mark it as a task.
- This decorator accepts a number of optional arguments. For details, see Task-level config.
We call app.start() in our code's entry point.
- On Render, this is what kicks off the task registration process and the execution of each run.

Organizing tasks

You can define your workflow's tasks across multiple files in your project repo:

math_tasks.py

python

from render_sdk import Workflows

app = Workflows()

@app.task
def add(a: int, b: int) -> int:
  return a + b

text_tasks.py

python

from render_sdk import Workflows

app = Workflows()

@app.task
def capitalize(s: str) -> str:
  return s.upper()

main.py

python

from render_sdk 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() # SDK entry point

In this example, task definitions are distributed across two files: math_tasks.py and text_tasks.py.

To register all of your tasks, your workflow's entry point (commonly main.py) imports and incorporates the Workflows apps from each other file using the Workflows.from_workflows() method.

Task arguments

A task function can define any number of arguments. This example task takes three arguments of different types:

python

@app.task
def my_task(arg1: int, arg2: str, arg3: bool) -> int:
  # ...

Argument and return types must be JSON-serializable. Your applications provide task arguments in a JSON array or object via the Render API. A task's result is also returned as JSON.

Task arguments are required by default. However, the Render SDK for Python does support setting default argument values:

python

@app.task
def my_task(arg1: int = 3) -> int:
  # ...

If you attempt to trigger a run with missing arguments (or too many arguments), the run will fail. Argument values can be null, as long as your function's logic supports this.

Task-level config

Instance type (compute specs)

By default, task runs execute on Render's Standard instance type (1 CPU, 2GB RAM). You can override this on a per-task basis.

Set a task's instance type with the following syntax:

python

@app.task(
  plan="starter"
)

The following instance types are supported for all workspaces:

Instance Type	Specs
`starter`	0.5 CPU 512 MB RAM
`standard` (default)	1 CPU 2 GB RAM
`pro`	2 CPU 4 GB RAM

If you have more resource-intensive workloads, you can request access to the following larger instance types:

Instance Type	Specs
`pro_plus`	4 CPU 8 GB RAM
`pro_max`	8 CPU 16 GB RAM
`pro_ultra`	16 CPU 32 GB RAM

See pricing details.

Timeout

By default, task runs time out after 2 hours. You can override this on a per-task basis to any value between 30 seconds and 24 hours.

Provide your task's timeout to the @app.task decorator via the timeout_seconds argument:

python

@app.task(
  timeout_seconds=86400 # 24 hours in seconds
)

Retry logic

Task runs can automatically retry if they fail. A run is considered to have failed if its function raises an exception instead of returning a value.

Retries are useful for tasks that might be affected by transient failures, such as network errors or timeouts.

Default retry behavior

By default, task runs use the following retry logic:

Retry up to 3 times (i.e., 4 total attempts)
Wait 1 second before attempting the first retry
Double the wait time after each retry (i.e., one second, two seconds, four seconds)

Customizing retries

You can customize retry behavior on a per-task basis. Every run of a task uses the same retry settings.

Provide retry settings to the @app.task decorator with the following syntax:

python

from render_sdk import Workflows, Retry
import random

app = Workflows()

@app.task(
  retry=Retry(
    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 (1s, 1.5s, 2.25s)
  )
)
def flip_coin() -> str:
  if random.random() < 0.5:
    raise Exception("Flipped tails! Retrying.")
  return "Flipped heads!"

This contrived example defines a task named flip_coin that raises an exception when it "flips tails", causing the run to fail and retry according to its settings.

Chaining task runs

A task run can trigger additional task runs. Like other runs, these chained runs each execute in their own instance.

When should I chain runs?

Chaining is most helpful when different parts of a larger job benefit from long-running, individually provisioned compute.

For simple jobs (such as the very basic example below), it's more efficient to define the entirety of your logic in a single task.

Example

The simple sum_squares task below chains two parallel runs of the calculate_square task:

math_tasks.py

python

from render_sdk import Workflows
import asyncio

app = Workflows()

# A task that chains two parallel runs
@app.task
async def sum_squares(a: int, b: int) -> int: # Must be async to await chained runs
  result1, result2 = await asyncio.gather(
    calculate_square(a),
    calculate_square(b)
  )
  return result1 + result2

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

When chaining runs:

In most cases, your chaining task's function should be defined as async.
- Otherwise, it can't await the results of its chained runs.
You chain a run by calling the corresponding task function (e.g., calculate_square above).
- However, this call doesn't return the function's defined return value!
- Instead, this triggers a new run and returns a special TaskInstance object.
- As shown, you can await this object to obtain the run's actual return value.

Task functions can call other functions that are not marked as tasks. These functions execute and return as normal (they do not trigger chained runs).

Need to run a task defined in a different workflow?

This requires instead using the Render SDK or Render API, as described in Running Workflow Tasks. Note that this is not tracked as a chaining relationship when visualizing task execution in the Render Dashboard.

Parallel runs

When chaining runs, you'll often want to kick off multiple in parallel, such as to chunk a large workload into smaller independent pieces. Common examples include processing batches of images or analyzing different sections of a large document.

To chain parallel runs in Python, use asyncio.gather, asyncio.TaskGroup, or a similar concurrency utility.

In this example, the process_photo_upload task chains a separate process_image run for each element in its image_urls argument:

python

from render_sdk import Workflows
import asyncio

app = Workflows()

@app.task
async def process_photo_upload(image_urls: list[str]) -> dict:
  # Process all images in parallel by chaining a run for each
  results = await asyncio.gather( 
    *[process_image(url) for url in image_urls] 
  ) 

  num_successful = sum(1 for r in results if r["success"])
  num_failed = len(results) - num_successful

  return {
    "total": len(image_urls),
    "processed": num_successful,
    "failed": num_failed,
    "results": results
  }

@app.task
def process_image(image_url: str) -> dict:

  # Image processing logic goes here

  return {
    "url": image_url,
    "thumbnail_url": f"{image_url}_thumb.jpg",
    "success": True
  }

If you don't use asyncio.gather or a similar function, chained runs execute serially.

For example:

python

@app.task
async def sum_squares_slower(a: int, b: int) -> int:
  # ⚠️ Not parallel!
  result1 = await calculate_square(a)
  result2 = await calculate_square(b) # Executes after first run completes
  return result1 + result2

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

Serial execution is helpful when one run depends on the result of another. However, it can significantly slow execution for runs that are completely independent. Parallelize wherever your use case allows.