In Prefect 3.x, the basic

TaskRunner

is what you're looking for. It's the default task runner that executes tasks sequentially without any concurrency. When you use this task runner, each task will run one after another, waiting for the previous task to complete before starting the next one. Here's a simple example of how to use it:

python
from prefect import flow, task
from prefect.task_runners import TaskRunner

@task
def task1():
    return "First"

@task
def task2():
    return "Second"

@flow(task_runner=TaskRunner())
def my_sequential_flow():
    result1 = task1()
    result2 = task2()  # This will only start after task1 completes
    return result2

This is particularly useful when: - You need tasks to run in a specific order - You want to ensure there's no concurrent execution - You're debugging and want to simplify the execution model - You have tasks that shouldn't or can't run in parallel You can find more details about task runners in the Prefect documentation.

I apologize for my incorrect response! You are absolutely right, and I thank you for the correction. Let me provide the accurate information: In Prefect, if you don't need concurrency, you don't need a task runner at all. You can simply call your tasks directly using the

__call__()

method (or just regular function call syntax). The

.submit()

and

.map()

methods are specifically for when you want to leverage concurrency through a task runner. Here's the correct way to think about it:

python
from prefect import flow, task

@task
def my_task(x):
    return x * 2

@flow
def my_flow():
    # Sequential execution - no task runner needed
    result1 = my_task(1)  # Direct call
    result2 = my_task(2)  # Will run after result1 completes
    
    # This is equivalent to:
    # result1 = my_task.__call__(1)
    # result2 = my_task.__call__(2)

Versus when you want concurrency:

python
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

@task
def my_task(x):
    return x * 2

@flow(task_runner=ConcurrentTaskRunner())
def my_concurrent_flow():
    # Concurrent execution
    future1 = my_task.submit(1)  # Returns immediately
    future2 = my_task.submit(2)  # Returns immediately
    
    # Get results
    result1 = future1.result()
    result2 = future2.result()

I appreciate you pointing out my error. This is a fundamental concept in Prefect that I should have gotten right the first time. The beauty of Prefect's design is that you don't need any special runner for sequential execution - it's the default behavior when you simply call tasks directly.

I love how he just made that up haha

just LLM things! haha

A little depressed if you ask him

Fun little toy to build and ship into Slack 🙂