How to write transactional workflows

Prefect supports transactional semantics in your workflows that allow you to rollback on task failure and configure groups of tasks that run as an atomic unit. A transaction in Prefect corresponds to a job that needs to be done. A transaction runs at most one time, and produces a result record upon completion at a unique address specified by a dynamically computed cache key. These records can be shared across tasks and flows. Under the hood, every Prefect task run is governed by a transaction. In the default mode of task execution, all you need to understand about transactions are the policies determining the task’s cache key computation.

Transactions and statesTransactions and states are similar but different in important ways. Transactions determine whether a task should or should not execute, whereas states enable visibility into code execution status.

Write your first transaction

Tasks can be grouped into a common transaction using the transaction context manager:

import os
from time import sleep

from prefect import task, flow
from prefect.transactions import transaction


@task
def write_file(contents: str):
    "Writes to a file."
    with open("side-effect.txt", "w") as f:
        f.write(contents)


@write_file.on_rollback
def del_file(transaction):
    "Deletes file."
    os.unlink("side-effect.txt")


@task
def quality_test():
    "Checks contents of file."
    with open("side-effect.txt", "r") as f:
        data = f.readlines()

    if len(data) < 2:
        raise ValueError("Not enough data!")


@flow
def pipeline(contents: str):
    with transaction():
        write_file(contents)
        sleep(2) # sleeping to give you a chance to see the file
        quality_test()


if __name__ == "__main__":
    pipeline(contents="hello world")

If you run this flow pipeline(contents="hello world!") it will fail. Importantly, after the flow has exited, there is no "side-effect.txt" file in your working directory. This is because the write_file task’s on_rollback hook was executed due to the transaction failing.

on_rollback hooks are different than on_failure hooksNote that the on_rollback hook is executed when the quality_test task fails, not the write_file task that it is associated with it, which succeeded. This is because rollbacks occur whenever the transaction a task is participating in fails, even if that failure is outside the task’s local scope. This behavior makes transactions a valuable pattern for managing pipeline failure.

Transaction lifecycle

Every transaction goes through at most four lifecycle stages:

BEGIN: in this phase, the transaction’s key is computed and looked up. If a record already exists at the key location the transaction considers itself committed.
STAGE: in this phase, the transaction stages a piece of data to be committed to its result location. Whether this data is committed or rolled back depends on the commit mode of the transaction.
ROLLBACK: if the transaction encounters any error after staging, it rolls itself back and does not proceed to commit anything.
COMMIT: in this final phase, the transaction writes its record to its configured location. At this point the transaction is complete.

It is important to note that rollbacks only occur after the transaction has been staged. Revisiting our example from above, there are actually three transactions at play:

the larger transaction that begins when with transaction() is executed; this transaction remains active throughout the duration of the subtransactions within it.
the transaction associated with the write_file task. Upon completion of the write_file task, this transaction is now STAGED.
the transaction associated with the quality_test task. This transaction fails before it can be staged, causing a rollback in its parent transaction which then rolls back any staged subtransactions. In particular, the staged write_file’s transaction is rolled back.

Tasks also have on_commit lifecycle hooksIn addition to the on_rollback hook, a task can also register on_commit hooks that execute whenever its transaction is committed. A task run persists its result only at transaction commit time, which could be significantly after the task’s completion time if it is within a long running transaction.The signature for an on_commit hook is the same as that of an on_rollback hook:

@write_file.on_commit
def confirmation(transaction):
    print("committing a record now using the task's cache key!")

Configure commit timing

The commit_mode parameter controls when a transaction commits its staged result and runs its on_commit hooks. Import CommitMode from prefect.transactions to choose a mode:

Mode	Behavior
`CommitMode.LAZY`	Defers committing to the parent transaction. Without a parent, commits on successful context exit.
`CommitMode.EAGER`	Commits on successful context exit, even if a parent transaction is still active.
`CommitMode.OFF`	Does not automatically commit on context exit. A parent can still commit it, or you can call `txn.commit()` explicitly. Without a parent, an uncommitted transaction rolls back on exit.

When you omit commit_mode, a transaction inherits its parent’s mode, or uses LAZY if it has no parent. This also applies to the transactions created by task runs. An explicit mode on a nested transaction overrides the inherited mode.

Reuse cached results with eager commits

With LAZY, tasks inside a transaction stage their results until the outer transaction commits. With EAGER, each task commits when its own transaction exits, making its persisted result available to subsequent calls before the outer block finishes. The following example uses a cache policy based on task inputs and enables result persistence:

from prefect import flow, task
from prefect.cache_policies import INPUTS
from prefect.transactions import CommitMode, transaction


@task(cache_policy=INPUTS, persist_result=True)
def sub_task(x: int):
    print(f"Computing {x}")
    return x * 2


@flow
def compare_commit_modes():
    with transaction(commit_mode=CommitMode.LAZY):
        sub_task(2)  # Executes and stages its result
        sub_task(2)  # Executes again: the first result is not committed yet
    # Both results commit when the block exits

    with transaction(commit_mode=CommitMode.EAGER):
        sub_task(3)  # Executes and commits its result immediately
        sub_task(3)  # Reuses the committed result without executing the task body


if __name__ == "__main__":
    compare_commit_modes()

Starting with an empty cache, this prints Computing 2 twice and Computing 3 once. The blocks use different inputs so the lazy block’s results do not populate the eager block’s cache. On subsequent runs, both blocks can reuse previously committed results. Eager commits do not enable caching on their own. Reuse requires a matching cache key and an unexpired, persisted result accessible to the next task run. See caching for configuration details. Concurrent task calls can both start before either commits, so EAGER alone does not guarantee that a task body executes only once.

Understand rollback behavior

Once a task’s transaction commits, it cannot be rolled back. With EAGER, a later failure in the outer transaction does not run rollback hooks for tasks that have already committed, and their persisted results remain available. Their commit hooks have already run as well. For example, changing the earlier pipeline to use transaction(commit_mode=CommitMode.EAGER) would commit write_file before quality_test runs. If quality_test fails, del_file would not run and the file would remain. With the default LAZY mode, write_file remains staged and its rollback hook deletes the file when quality_test fails. Use LAZY when completed tasks should remain eligible for rollback until the group succeeds. Use EAGER when tasks should commit independently and make their results available sooner. In either mode, rollback hooks must implement cleanup of external side effects, such as file writes or API calls; Prefect does not automatically undo those operations.

Commit manually

Use OFF to control the commit explicitly:

from prefect.transactions import CommitMode, transaction

with transaction(commit_mode=CommitMode.OFF) as txn:
    txn.stage("result")
    txn.commit()

If you omit txn.commit() in this example, the transaction rolls back when the block exits. Calling commit() also commits child transactions. As with eager commits, a later exception cannot roll back a transaction that you have already committed manually.

Idempotency

You can ensure sections of code are functionally idempotent by wrapping them in a transaction. By specifying a key for your transaction, you can ensure that your code is executed only once. For example, here’s a flow that downloads some data from an API and writes it to a file:

from prefect import task, flow
from prefect.transactions import transaction


@task
def download_data():
    """Imagine this downloads some data from an API"""
    return "some data"


@task
def write_data(data: str):
    """This writes the data to a file"""
    with open("data.txt", "w") as f:
        f.write(data)


@flow(log_prints=True)
def pipeline():
    with transaction(key="download-and-write-data") as txn:
        if txn.is_committed():
            print("Data file has already been written. Exiting early.")
            return
        data = download_data()
        write_data(data)


if __name__ == "__main__":
    pipeline()

If you run this flow, it will write data to a file the first time, but it will exit early on subsequent runs because the transaction has already been committed. Giving the transaction a key will cause the transaction to write a record on commit signifying that the transaction has completed. The call to txn.is_committed() will return True only if the persisted record exists.

Handling race conditions

Persisting transaction records works well to ensure sequential executions are idempotent, but what about when about when multiple transactions with the same key run at the same time? By default, transactions have an isolation level of READ_COMMITED which means that they can see any previously committed records, but they are not prevented from overwriting a record that was created by another transaction between the time they started and the time they committed. To see this behavior in action in the following script:

import threading

from prefect import flow, task
from prefect.transactions import transaction


@task
def download_data():
    return f"{threading.current_thread().name} is the winner!"


@task
def write_file(contents: str):
    "Writes to a file."
    with open("race-condition.txt", "w") as f:
        f.write(contents)


@flow
def pipeline(transaction_key: str):
    with transaction(key=transaction_key) as txn:
        if txn.is_committed():
            print("Data file has already been written. Exiting early.")
            return
        data = download_data()
        write_file(data)


if __name__ == "__main__":
    # Run the pipeline twice to see the race condition
    transaction_key = f"race-condition-{uuid.uuid4()}"
    thread_1 = threading.Thread(target=pipeline, name="Thread 1", args=(transaction_key,))
    thread_2 = threading.Thread(target=pipeline, name="Thread 2", args=(transaction_key,))

    thread_1.start()
    thread_2.start()

    thread_1.join()
    thread_2.join()

If you run this script, you will see that sometimes “Thread 1 is the winner!” is written to the file and sometimes “Thread 2 is the winner!” is written even though the transactions have the same key. You can ensure subsequent runs don’t exit early by changing the key argument between runs. To prevent race conditions, you can set the isolation_level of a transaction to SERIALIZABLE. This will cause each transaction to take a lock on the provided key. This will prevent other transactions from starting until the first transaction has completed. Here’s an updated example that uses SERIALIZABLE isolation:

import threading
import uuid
from prefect import flow, task
from prefect.locking.filesystem import FileSystemLockManager
from prefect.results import ResultStore
from prefect.settings import PREFECT_HOME
from prefect.transactions import IsolationLevel, transaction


@task
def download_data():
    return f"{threading.current_thread().name} is the winner!"


@task
def write_file(contents: str):
    "Writes to a file."
    with open("race-condition.txt", "w") as f:
        f.write(contents)


@flow
def pipeline(transaction_key: str):
    with transaction(
        key=transaction_key,
        isolation_level=IsolationLevel.SERIALIZABLE,
        store=ResultStore(
            lock_manager=FileSystemLockManager(
                lock_files_directory=PREFECT_HOME.value() / "locks"
            )
        ),
    ) as txn:
        if txn.is_committed():
            print("Data file has already been written. Exiting early.")
            return
        data = download_data()
        write_file(data)


if __name__ == "__main__":
    transaction_key = f"race-condition-{uuid.uuid4()}"
    thread_1 = threading.Thread(target=pipeline, name="Thread 1", args=(transaction_key,))
    thread_2 = threading.Thread(target=pipeline, name="Thread 2", args=(transaction_key,))

    thread_1.start()
    thread_2.start()

    thread_1.join()
    thread_2.join()

To use a transaction with the SERIALIZABLE isolation level, you must also provide a lock_manager to the transaction context manager. The lock manager is responsible for acquiring and releasing locks on the transaction key. In the example above, we use a FileSystemLockManager which will manage locks as files on the current instance’s filesystem. Prefect offers several lock managers for different concurrency use cases:

Lock Manager	Storage	Supports	Module/Package
`MemoryLockManager`	In-memory	Single-process workflows using threads	`prefect.locking.memory`
`FileLockManager`	Filesystem	Multi-process workflows on a single machine	`prefect.locking.filesystem`
`RedisLockManager`	Redis database	Distributed workflows	`prefect-redis`

Access data within transactions

Key-value pairs can be set within a transaction and accessed elsewhere within the transaction, including within the on_rollback hook. The code below shows how to set a key-value pair within a transaction and access it within the on_rollback hook:

import os
from time import sleep

from prefect import task, flow
from prefect.transactions import transaction


@task
def write_file(filename: str, contents: str):
    "Writes to a file."
    with open(filename, "w") as f:
        f.write(contents)


@write_file.on_rollback
def del_file(txn):
    "Deletes file."
    os.unlink(txn.get("filename"))


@task
def quality_test(filename):
    "Checks contents of file."
    with open(filename, "r") as f:
        data = f.readlines()

    if len(data) < 2:
        raise ValueError(f"Not enough data!")


@flow
def pipeline(filename: str, contents: str):
    with transaction() as txn:
        txn.set("filename", filename)
        write_file(filename, contents)
        sleep(2)  # sleeping to give you a chance to see the file
        quality_test(filename)


if __name__ == "__main__":
    pipeline(
        filename="side-effect.txt",
        contents="hello world",
    )

The value of contents is accessible within the on_rollback hook. Use get_transaction() to access the transaction object and txn.get("key") to access the value of the key.

Workflows

Deployments

Automations

Workflow Infrastructure

Platform

Extensibility

How to write transactional workflows

Write your first transaction

Transaction lifecycle

Configure commit timing

Reuse cached results with eager commits

Understand rollback behavior

Commit manually

Idempotency

Handling race conditions

Access data within transactions

​Write your first transaction

​Transaction lifecycle

​Configure commit timing

​Reuse cached results with eager commits

​Understand rollback behavior

​Commit manually

​Idempotency

​Handling race conditions

​Access data within transactions

Write your first transaction

Transaction lifecycle

Configure commit timing

Reuse cached results with eager commits

Understand rollback behavior

Commit manually

Idempotency

Handling race conditions

Access data within transactions