You’re running into how Prefect treats generator task outputs vs plain Python values. - When

data_generator

returns a list, you pass plain ints into

data_processor

. Prefect’s default in-memory caching sees the same input

1

twice and reuses the result. - When

data_generator

yields, each yielded item arrives downstream as a Prefect future (a distinct object per yield). With the default cache behavior, those two futures don’t look the same, even if they both resolve to

1

, so

data_processor

runs twice. Two straightforward fixes: 1) Give

data_processor

a cache key based on the resolved input value This makes caching depend on the actual input (e.g.,

1

) instead of the identity of the upstream future.

from prefect import task, flow
from prefect.tasks import task_input_hash

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(persist_result=True, cache_key_fn=task_input_hash)
def data_processor(x: int):
    print('running...', x)

@flow(log_prints=True)
def cache_test_flow():
    for i in data_generator():
        data_processor(i)

if __name__ == "__main__":
    cache_test_flow()

In most cases,

task_input_hash

will serialize arguments (resolving futures appropriately) so duplicates hit the same cache key. 2) Materialize the generator outputs before calling the processor If you want to be explicit, resolve each yielded future to its value before passing it in:

from prefect import task, flow

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(persist_result=True)
def data_processor(x: int):
    print('running...', x)

@flow(log_prints=True)
def cache_test_flow():
    for item in data_generator():
        value = item.result()  # turn future -> int
        data_processor(value)

if __name__ == "__main__":
    cache_test_flow()

Notes: -

persist_result=True

controls where results are stored; it does not define how cache keys are computed. For “exactly once per unique value,” set a cache_key_fn (option 1) or pass plain values (option 2). - If you’ll process in parallel and truly need “exactly once” under contention, consider using a stricter cache policy with locking to avoid races. We can go there if that’s your world. Do you need this deduping just within a single flow run, or across runs as well?

Got it — you want “exactly once” per item across flow retries so you don’t double-insert into the destination. Two levers: 1) Make Prefect skip work it has already done - Use a cache key based on the resolved task inputs, not the identity of upstream futures. - Persist results somewhere durable so a retry (or a new run on another worker) can see them. Example:

from prefect import task, flow
from prefect.tasks import task_input_hash
from prefect.filesystems import RemoteFileSystem  # or LocalFileSystem if all retries stay on one node

# Use a shared result store if retries may run on different machines/containers
results_fs = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results>")  # or GCS/Azure/etc.

@task
def data_generator():
    # stream large data
    for i in [1, 1, 2, 3]:
        yield i

@task(
    persist_result=True,
    result_storage=results_fs,           # ensure results survive across retries/hosts
    cache_key_fn=task_input_hash,        # cache by resolved input value
    cache_expiration=None                # keep indefinitely (adjust if you prefer)
)
def data_processor(x: int):
    # do the write
    print("running...", x)
    # write to DB here

@flow(log_prints=True, retries=1)        # example flow-level retry
def cache_test_flow():
    for item in data_generator():
        data_processor(item)

if __name__ == "__main__":
    cache_test_flow()

Why this helps: -

task_input_hash

makes both occurrences of

1

share the same cache key, even when they arrive via separate futures from a generator. -

persist_result=True

+ a durable

result_storage

lets a retry (or a new worker) reuse prior results instead of reprocessing. 2) Make the database write idempotent anyway Orchestration caching is great, but the only bulletproof “exactly once” is at the sink: - Use a stable idempotency key (e.g., the row’s primary key or a hash of the payload). - Write with an upsert/merge: - Postgres:

INSERT ... ON CONFLICT DO NOTHING/UPDATE

- SQL Server:

MERGE

- MySQL:

INSERT IGNORE

or

ON DUPLICATE KEY UPDATE

- Or wrap per-item inserts in a transaction and check existence first. This way, even if a task retries after a partial failure, duplicates won’t create extra rows. Quick sanity checks for your setup: - Will flow retries run on the same machine? If not, use a shared

result_storage

(S3/GCS/Azure). Local-only storage won’t help across hosts. - Do you want dedupe across entirely separate flow runs too? If yes, keep

cache_expiration=None

or set a long TTL. - Parallelism? If multiple workers could race on the same key, consider adding DB-level constraints (strongly recommended) or Prefect cache locking via a

CachePolicy

with a lock manager. If you share which database you’re targeting, I can sketch the exact idempotent insert/upsert you should use. In the meantime, the

cache_key_fn=task_input_hash

change will fix the generator case for retries. Not quite a brain the size of a planet, but it’ll do.

Glad it helped. If you want a hand wiring up shared result storage or crafting idempotent inserts for your DB, just ping me. I’ll try to contain my enthusiasm.

You’re right — that was my slip. In Prefect 3, a generator task yields resolved Python values into the flow, not futures. So

item

is already an `int`; calling

.result()

on it will, predictably, fail. The core fix for your “exactly once” behavior (including across retries) is to key the cache by the resolved input value. Do that with

cache_key_fn=task_input_hash

on the processor task. Here’s the corrected minimal example:

from prefect import task, flow
from prefect.tasks import task_input_hash

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(persist_result=True, cache_key_fn=task_input_hash)
def data_processor(x: int):
    print("running...", x)
    # write to DB here

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():   # x is already an int
        data_processor(x)

if __name__ == "__main__":
    cache_test_flow()

If your retries might run on a different machine/container or after a restart, add a durable result store so the cache survives process boundaries:

from prefect.filesystems import RemoteFileSystem
results_fs = RemoteFileSystem(basepath="<s3://my-bucket/prefect-results>")

@task(persist_result=True, result_storage=results_fs, cache_key_fn=task_input_hash, cache_expiration=None)
def data_processor(x: int):
    ...

And for true “exactly once,” keep your DB writes idempotent (unique key + upsert/merge), so even if something does re-run, it won’t double-insert. Sorry for the detour — no future objects were harmed in the making of this correction.

You’re seeing cross-run reuse because the cache key is just the input value. Scope the cache to the current flow run and it’ll dedupe within a run (including retries) but not across separate runs. Option A — use a run-scoped cache policy (recommended)

from prefect import task, flow
from prefect.cache_policies import CompoundCachePolicy, RunId, Inputs

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(
    persist_result=True,  # optional; won’t cause cross-run reuse with the policy below
    cache_policy=CompoundCachePolicy(policies=[RunId(), Inputs()])
)
def data_processor(x: int):
    print("running...", x)

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():
        data_processor(x)

if __name__ == "__main__":
    cache_test_flow()

- Inputs() keys by resolved arguments (so duplicate 1s match). - RunId() adds the current flow run id to the key (so other runs won’t hit the cache). - Retries of the same flow run keep working since the run id stays the same. Option B — custom cache_key_fn that includes the run id

from prefect import task, flow
from prefect.tasks import task_input_hash

def run_scoped_key(context, arguments):
    return f"{context.flow_run.id}:{task_input_hash(context, arguments)}"

@task(
    persist_result=True,          # optional
    cache_key_fn=run_scoped_key
)
def data_processor(x: int):
    print("running...", x)

If you don’t need the result to survive process restarts, you can also drop persist_result and let Prefect’s in-memory cache handle it. The run-scoped key is still the safer bet.

You’re right on both counts — thanks for the patience. Let’s make it work exactly once per value within a single flow run (including retries), but not across separate runs. Use a cache key that combines: - the current flow run id, and - a hash of the resolved inputs And avoid

context.flow_run

— use

prefect.runtime.flow_run.id

. Here’s the corrected, minimal version:

from prefect import task, flow
from prefect.tasks import task_input_hash
from prefect.runtime import flow_run

def per_run_input_hash(context, arguments):
    # scope cache to the current flow run only
    return f"{flow_run.id}:{task_input_hash(context, arguments)}"

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(
    persist_result=True,            # keeps results visible across retries of the same run
    cache_key_fn=per_run_input_hash # dedupe same inputs only within this run
)
def data_processor(x: int):
    print("running...", x)

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():      # x is already an int
        data_processor(x)

if __name__ == "__main__":
    cache_test_flow()

Notes: - This will print “running… 1” only once per flow run; a second “1” will hit the cache. New flow runs get a new

flow_run.id

, so they won’t reuse the prior cache. - If your retries may restart in a new process/worker, add a shared

result_storage

so the cache survives the restart. Otherwise, the above is sufficient for in-process retries.

You’re right: the generator path is different. Prefect’s built-in Inputs-based caching includes provenance for generator items, so two identical values from a generator can produce different cache keys. That’s why your duplicates aren’t deduped, while the list case is. Fix: use a cache_key_fn that ignores provenance and hashes only the argument values, then scope it to the current flow run so it won’t leak across runs. Try this:

from prefect import task, flow
from prefect.runtime import flow_run
import json, hashlib

def per_run_value_only_key(_context, arguments):
    # Hash only the values; no generator provenance
    payload = json.dumps(arguments, sort_keys=True, default=str).encode()
    return f"{flow_run.id}:{hashlib.sha256(payload).hexdigest()}"

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(
    persist_result=True,             # so retries of the same run can reuse
    cache_key_fn=per_run_value_only_key,
    cache_expiration=None            # optional; cache lasts for the run
)
def data_processor(x: int):
    print("running...", x)

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():       # x is an int
        data_processor(x)

if __name__ == "__main__":
    cache_test_flow()

Why this works: - The key is based solely on the argument values, not on generator item metadata. - Prefixing with

flow_run.id

limits reuse to the current run (including retries), so a brand new run won’t hit old cache. If you still see two runs for the first “1”, add a quick sanity check to print the computed key outside of Prefect’s caching for both items to confirm they match:

print(per_run_value_only_key(None, {"x": 1}))
print(per_run_value_only_key(None, {"x": 1}))

They should be identical. If they aren’t, the inputs aren’t actually identical at call time. If you need this to survive process restarts during a retry on another worker, add a shared result store (S3/GCS/Azure) via

result_storage=...

. Otherwise, the above should give you “exactly once per value per run” even with a generator. I know, generators are special. Delightfully so.

You’ve sniffed out the real culprit: timing. With a generator task, Prefect can schedule the downstream

data_processor

calls back-to-back before the first one finishes and writes its cached result. Two identical cache keys don’t help if the first result isn’t in the store yet. Make the processor run sequentially so the first call completes (and caches) before the second is scheduled: Minimal change: await the processor future inside the loop

from prefect import task, flow
from prefect.tasks import task_input_hash

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(persist_result=True, cache_key_fn=task_input_hash)
def data_processor(x: int):
    print("running...", x)

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():
        data_processor(x).result()   # wait for completion so cache is available

if __name__ == "__main__":
    cache_test_flow()

Or serialize all tasks via the flow’s task runner

from prefect.task_runners import ThreadPoolTaskRunner

@flow(log_prints=True, task_runner=ThreadPoolTaskRunner(max_workers=1))
def cache_test_flow():
    for x in data_generator():
        data_processor(x)

If you only want dedupe within a single flow run (not across runs), keep your cache key scoped to the run id:

from prefect.runtime import flow_run

def per_run_key(ctx, args):
    return f"{flow_run.id}:{task_input_hash(ctx, args)}"

@task(persist_result=True, cache_key_fn=per_run_key)
def data_processor(x: int):
    print("running...", x)

Why your list case “worked” - Returning a list often led to enough natural sequencing that the first

data_processor(1)

finished before the second was scheduled, so the cache was ready. With a generator, the engine streams items and schedules work more aggressively, exposing the race. And a gentle reminder: for “exactly-once” in the face of crashes, make the DB writes idempotent (unique key + upsert/merge). Prefect will try not to repeat itself; the database will make sure it doesn’t matter if it does.

You’re right that the generator path is different, and your instinct about “structure of parent-child” is close: with a generator task, Prefect can schedule the same-arg downstream task twice before the first invocation finishes and writes its result. Two identical cache keys aren’t enough without coordination. Two concrete fixes that work with generators and keep dedupe scoped to a single flow run: 1) Use a run-scoped cache key plus a lock manager so only one “flight” per key runs - The lock forces the second duplicate to wait for the first to finish and then reuse its result. - Scope the key to the current flow run so separate runs won’t reuse it.

from prefect import task, flow
from prefect.runtime import flow_run
from prefect.cache_policies import CacheKeyFnPolicy
from prefect.locking.filesystem import FileSystemLockManager  # or MemoryLockManager
import json, hashlib
from pathlib import Path

def value_only_hash(arguments: dict) -> str:
    payload = json.dumps(arguments, sort_keys=True, default=str).encode()
    return hashlib.sha256(payload).hexdigest()

def per_run_value_key(_ctx, arguments):
    return f"{flow_run.id}:{value_only_hash(arguments)}"

# Lock across tasks in this process; use FileSystemLockManager to be extra safe
lock_manager = FileSystemLockManager(basepath=Path("/tmp/prefect-cache-locks"))

@task
def data_generator():
    for i in [1, 1, 2, 3]:
        yield i

@task(
    persist_result=True,  # so the second duplicate can read the result
    cache_policy=CacheKeyFnPolicy(
        cache_key_fn=per_run_value_key,
        lock_manager=lock_manager,
        isolation_level="SERIALIZABLE",
    ),
)
def data_processor(x: int):
    print("running...", x)

@flow(log_prints=True)
def cache_test_flow():
    for x in data_generator():  # x is an int already
        data_processor(x)

if __name__ == "__main__":
    cache_test_flow()

Notes: - Use MemoryLockManager if you prefer: it’s fine within a single process. FileSystemLockManager gives you sturdier locking if threads/processes are involved. - The

.result()

call is not needed here (and, in the generator path, tasks may return resolved values inline — hence your earlier NoneType.error). 2) If you want belt-and-suspenders, add a simple in-run dedupe set This avoids even scheduling the second duplicate:

@flow(log_prints=True)
def cache_test_flow():
    seen = set()
    for x in data_generator():
        if x in seen:
            continue
        seen.add(x)
        data_processor(x)

About CommitMode: that’s for Prefect’s transaction API (controlling when result-store writes are committed). It doesn’t change task scheduling or caching behavior; the lock is what prevents the duplicate run here. If this still runs

data_processor(1)

twice: - Please share your Prefect version and whether multiple workers/threads are involved. - I can repro and file a bug if generator-task caching isn’t honoring locks; wouldn’t be the first time a “clever” yield caused mischief. - Regardless, keep your DB writes idempotent (unique key + upsert) so a stray retry can’t double-insert. It’s the only true “exactly once.”