Yes — the usual pattern is: let fetches run concurrently, then serialize only the append/write section. In Prefect 3.x, I’d think about this in two layers: 1. Prefect/orchestration layer → controls who is allowed into the critical section 2. Data/storage layer → ensures the actual append/write is safe and atomic You generally want both. --- ## 1) Establish the lock in the Prefect layer For orchestration-level locking, use a global concurrency limit with a limit of

1

. Prefect provides

prefect.concurrency.sync.concurrency(...)

for this, and the CLI confirms the

gcl

commands exist. ### Create the lock

bash
prefect gcl create serialized-append --limit 1

CLI help confirms: -

prefect gcl create

-

prefect gcl inspect

-

prefect gcl ls

So you can inspect it with:

bash
prefect gcl inspect serialized-append

### Use it in the write section

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def fetch_record(i: int) -> str:
    # runs concurrently
    return f"record-{i}\n"

@task
def append_record(path: str, record: str):
    # only one task at a time may enter this block
    with concurrency("serialized-append", occupy=1, strict=True):
        with open(path, "a") as f:
            f.write(record)

@flow
def pipeline():
    fetched = [fetch_record.submit(i) for i in range(20)]
    writes = [append_record.submit("out.txt", fut.result()) for fut in fetched]
    [w.result() for w in writes]

### What this gives you - all

fetch_record

tasks can run in parallel - only one

append_record

task can execute the guarded block at a time - this works across distributed workers, as long as they share the same Prefect API/backend ### Important caveat This is an orchestration lock, not a filesystem/database transaction. It prevents Prefect-managed code paths from entering the critical section concurrently, but it does not by itself make the underlying storage append atomic. So if correctness matters, add a data-layer lock/transaction too. --- ## 2) Apply the lock in the data layer This depends on what “serialized append” means in your system. ## Best practice Use Prefect concurrency to guard the critical section and use a storage-native or transaction-based mechanism for the actual write: - DB table append → database transaction / row lock / advisory lock - object store append emulation → write temp object + compose/merge strategy - local/shared file append → OS/file lock - distributed multi-machine lock → Redis or another distributed lock service --- ## 3) Prefect-native transaction/lock support Prefect 3.x also exposes transaction primitives: -

prefect.transactions.transaction

-

prefect.transactions.IsolationLevel

- lock managers such as: -

prefect.locking.memory.MemoryLockManager

-

prefect.locking.filesystem.FileSystemLockManager

Those are useful when you want Prefect-managed transaction semantics around a critical section. ### Example with

transaction(...)

and filesystem lock manager

python
from pathlib import Path

from prefect import flow, task
from prefect.transactions import transaction, IsolationLevel
from prefect.locking.filesystem import FileSystemLockManager
from prefect.results import ResultStore

lock_manager = FileSystemLockManager(
    lock_files_directory=Path(".prefect-locks")
)

store = ResultStore(lock_manager=lock_manager)

@task
def fetch_record(i: int) -> str:
    return f"record-{i}\n"

@task
def append_record(path: str, record: str):
    with transaction(
        key=f"append-{path}",
        store=store,
        isolation_level=IsolationLevel.SERIALIZABLE,
    ):
        with open(path, "a") as f:
            f.write(record)

@flow
def pipeline():
    fetched = [fetch_record.submit(i) for i in range(20)]
    writes = [append_record.submit("out.txt", fut.result()) for fut in fetched]
    [w.result() for w in writes]

### When to use this This is reasonable when:

- workers share a filesystem - you want Prefect-managed lock semantics - your write target is effectively local/shared-file based ### When not to rely on this alone If your workers are on different machines/containers without shared lock visibility,

FileSystemLockManager

is not enough. --- ## 4) Recommended pattern: Prefect lock + storage-native lock If your append target is a database, this is usually the strongest approach. ### Example: Prefect lock outside, DB transaction inside

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency
import sqlite3

@task
def fetch_record(i: int) -> tuple[int, str]:
    return (i, f"record-{i}")

@task
def append_record(db_path: str, row: tuple[int, str]):
    with concurrency("serialized-append", occupy=1, strict=True):
        conn = sqlite3.connect(db_path)
        try:
            conn.execute("BEGIN IMMEDIATE")
            conn.execute(
                "INSERT INTO records (id, value) VALUES (?, ?)",
                row,
            )
            conn.commit()
        finally:
            conn.close()

@flow
def pipeline():
    fetched = [fetch_record.submit(i) for i in range(20)]
    writes = [append_record.submit("data.db", fut.result()) for fut in fetched]
    [w.result() for w in writes]

This gives you: - Prefect layer: only one append task enters at a time - DB layer: transaction guarantees atomicity/durability That’s usually the safest design. --- ## 5) If you’re appending to a file File appends are tricky in distributed systems. If you are writing to: - local file on one machine: an OS/file lock can work - shared network filesystem: maybe, but reliability depends on the FS - S3/GCS/Azure blob: there is no real append lock; use object-versioned writes or write-per-task then merge ### Safer file pattern Instead of many tasks appending to one file, have each task write its own part:

python
@task
def write_part(i: int, record: str) -> str:
    path = f"parts/part-{i}.txt"
    with open(path, "w") as f:
        f.write(record)
    return path

@task
def merge_parts(paths: list[str], output_path: str):
    with open(output_path, "w") as out:
        for path in sorted(paths):
            with open(path) as part:
                out.write(part.read())

This often avoids the need for a distributed append lock entirely. --- ## 6) Which layer should own the lock? Short answer: - Prefect layer should own workflow concurrency policy - Data layer should own correctness of the write So: ### Establish lock in Prefect layer when: - you want to limit entry to the critical section - you need coordination across flow/task runs - you want simple serialization with

gcl --limit 1

### Establish lock in data layer when: - correctness matters even if something bypasses Prefect - the storage system has its own transaction/locking mechanism - you need atomic commit semantics --- ## 7) My recommendation for your case If your workflow is concurrent fetch, serialized append: ### Preferred design 1. fetch tasks run fully concurrently 2. append task uses a Prefect global concurrency limit with limit

1

3. inside the append task, use the data store’s own transaction/lock primitive That looks like:

python
from prefect import flow, task
from prefect.concurrency.sync import concurrency

@task
def fetch(i):
    return f"record-{i}\n"

@task
def append_safely(record, writer):
    with concurrency("serialized-append", strict=True):
        writer(record)

@flow
def run(writer):
    futures = [fetch.submit(i) for i in range(100)]
    writes = [append_safely.submit(f.result(), writer) for f in futures]
    [w.result() for w in writes]

Where

writer(...)

is implemented using your storage backend’s safe write semantics. --- ## 8) If you tell me your storage backend, I can make this concrete The right data-layer lock depends completely on where you append: - Postgres - Snowflake - local file

- NFS/shared volume - S3 - Delta/Iceberg/Hudi - Redis - something else If you want, reply with the write target and I can give you a production-ready pattern for: - how to establish the Prefect lock - how to implement the storage-layer append safely ## Relevant references - Prefect concurrency APIs:

prefect.concurrency.sync.concurrency

- Prefect transactions:

prefect.transactions.transaction

- Prefect lock managers:

prefect.locking.filesystem.FileSystemLockManager

- Prefect CLI for global concurrency limits:

prefect gcl ...

Docs/app links: - Prefect docs - Prefect Cloud