Yes—use a quick S3 pre-check and branch your flow so the downstream tasks aren’t called when the path exists. Here are two simple patterns in Prefect 3.x: 1) Check with the function API (one-off credentials)

from prefect import flow, task
from prefect_aws import AwsCredentials
from prefect_aws.s3 import list_objects

@task
def s3_path_exists(bucket: str, prefix: str) -> bool:
    creds = AwsCredentials.load("my-aws-creds")
    objs = list_objects(
        bucket=bucket,
        aws_credentials=creds,
        prefix=prefix,
        max_items=1,  # fast existence check
    )
    return bool(objs)

@task
def do_work():
    # your expensive logic here
    return "done"

@flow
def main(bucket: str, prefix: str):
    if s3_path_exists(bucket, prefix):
        print(f"Found s3://{bucket}/{prefix} — skipping downstream work.")
        return {"skipped": True}
    return {"result": do_work()}

2) Check with a pre-configured S3Bucket block

from prefect import flow, task
from prefect_aws.s3 import S3Bucket

@task
def s3_path_exists_via_block(bucket_block_name: str, prefix: str) -> bool:
    s3 = S3Bucket.load(bucket_block_name)
    return bool(s3.list_objects(folder=prefix, max_items=1))

@task
def do_work():
    return "done"

@flow
def main(bucket_block_name: str, prefix: str):
    if s3_path_exists_via_block(bucket_block_name, prefix):
        print(f"Found s3://<bucket>/{prefix} — skipping downstream work.")
        return {"skipped": True}
    return {"result": do_work()}

Notes: - Prefect 3.x doesn’t have a special “Skipped” state; the recommended approach is to use normal Python control flow (if condition) and not call the tasks at all. - If your goal is to prevent re-running the same task with the same inputs over time (e.g., output already exists in S3), you can also add caching to the task so repeated calls return a cached result instead of executing:

from datetime import timedelta
from prefect import task
from prefect.cache_policies import INPUTS, TASK_SOURCE

@task(cache_policy=INPUTS + TASK_SOURCE, cache_expiration=timedelta(hours=1))
def do_work(param: str):
    # runs once per unique param (and code), then returns cached result for 1h
    ...

Quick question to tailor this: do you want to skip the entire flow run if the S3 path exists, or just prevent specific tasks from running? And are you checking for a single key or any object under a prefix?

Great question. In Prefect 3.x, the task runner is set per-flow, not per-task. To mix Ray for some tasks and a thread pool for others, put those parts into separate subflows and give each subflow its own task runner. Example: main flow uses threads; a subflow uses Ray

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect_ray.task_runners import RayTaskRunner

@task
def cheap_task(x: int) -> int:
    # light work — better on threads
    return x + 1

@task
def heavy_task(x: int) -> int:
    # CPU-heavy or many-parallel — better on Ray
    return x * x

@flow(name="ray-stage", task_runner=RayTaskRunner(init_kwargs={"num_cpus": 4}))
def ray_stage(items: list[int]) -> list[int]:
    futures = [heavy_task.submit(i) for i in items]
    return [f.result() for f in futures]

@flow(name="main", task_runner=ThreadPoolTaskRunner(max_workers=8))
def main():
    # These run on a thread pool
    cheap_fs = [cheap_task.submit(i) for i in range(10)]
    cheap_results = [f.result() for f in cheap_fs]

    # This subflow (and its tasks) run on Ray
    ray_results = ray_stage(items=list(range(100)))
    return {"cheap": cheap_results, "ray": ray_results}

if __name__ == "__main__":
    main()

You can also flip it: make the outer flow use Ray and isolate a threads-only section in a ThreadPool subflow:

@flow(task_runner=ThreadPoolTaskRunner(max_workers=16))
def thread_stage(items: list[int]) -> list[int]:
    fs = [cheap_task.submit(i) for i in items]
    return [f.result() for f in fs]

@flow(task_runner=RayTaskRunner(init_kwargs={"num_cpus": 8}))
def ray_main(items: list[int]):
    # Ray-parallel work
    ray_fs = [heavy_task.submit(i) for i in items]
    ray_results = [f.result() for f in ray_fs]

    # Threads-only section
    small_results = thread_stage(items=list(range(20)))
    return {"ray": ray_results, "threads": small_results}

Notes: - Per-task runner selection isn’t supported; subflows are the clean way to mix executors. - To use Ray, install the integration:

pip install "prefect[ray]" ray

- If you already have a Ray cluster, pass `RayTaskRunner(address="auto")`; otherwise Prefect will start a local Ray runtime with

init_kwargs

. Does this setup fit your use case? Do you want the outer flow on threads or Ray, and are you connecting to an existing Ray cluster or running Ray locally?

Short answer: - Don’t use asyncio.gather on Prefect futures directly (they’re not awaitable). - Use resolve_futures_to_results or wait/as_completed for ThreadPool/ProcessPool/Ray futures. - Only PrefectTaskRunner futures support await via result_async()/wait_async(); then you can use asyncio.gather on those coroutines. Async example mixing ThreadPool (cheap) and Ray (heavy) via subflows

import asyncio
from prefect import flow, task
from prefect.futures import resolve_futures_to_results, as_completed
from prefect.task_runners import ThreadPoolTaskRunner
from prefect_ray.task_runners import RayTaskRunner

# Cheap async task — keep on threads
@task
async def cheap_task(i: int) -> str:
    await asyncio.sleep(0.1)
    return f"cheap-{i}"

# Heavy sync task — run on Ray
@task
def heavy_task(i: int) -> int:
    return sum(j*j for j in range(i * 10000))  # CPU-y example

# Ray subflow: runs heavy_task on Ray
@flow(name="ray-stage", task_runner=RayTaskRunner(init_kwargs={"num_cpus": 4}))
async def ray_stage(items: list[int]) -> list[int]:
    fs = [heavy_task.submit(i) for i in items]
    # Ray futures are not awaitable; use .result() or prefect.futures utilities
    return [f.result() for f in fs]  # or resolve_futures_to_results(fs)

# Main flow: async + threads for cheap work; call Ray subflow for heavy work
@flow(name="main", task_runner=ThreadPoolTaskRunner(max_workers=8))
async def main():
    # Fan out cheap async tasks on a thread pool
    cheap_fs = [cheap_task.submit(i) for i in range(10)]

    # Preferred: resolve all results (blocking call inside the async flow)
    cheap_results = resolve_futures_to_results(cheap_fs)

    # Or process incrementally as tasks complete
    for fut in as_completed(cheap_fs):
        _ = fut.result()  # do something incrementally

    # Hand heavy parts to Ray subflow (await the subflow itself)
    ray_results = await ray_stage(items=list(range(25)))

    return {"cheap": cheap_results, "ray": ray_results}

if __name__ == "__main__":
    asyncio.run(main())

When to use wait(), wait_async(), or asyncio.gather - ThreadPool/ProcessPool/Ray futures: - Use resolve_futures_to_results(futures) for the simplest “wait + unwrap” in both sync and async flows. - Use wait(futures) if you need (done, not_done) separation, or as_completed(futures) for streaming results. - There is no wait_async() for these futures; they are not awaitable. - PrefectTaskRunner futures (distributed, i.e., Prefect’s built-in runner): - You can stay fully async: await f.result_async() or await f.wait_async(). - You can also gather them concurrently:

# only for PrefectTaskRunner futures
    results = await asyncio.gather(*(f.result_async() for f in futures))

- Avoid asyncio.gather on Prefect futures directly (they are not awaitable). Only gather coroutines like f.result_async() from PrefectTaskRunner. Practical guidance - In most async flows, prefer resolve_futures_to_results; it’s simple and consistent across runners. - If you truly need non-blocking aggregation in an async flow with ThreadPool/Ray futures, either: - use as_completed to process incrementally, or - offload blocking resolution to a worker thread:

results = await asyncio.to_thread(resolve_futures_to_results, futures)

References - Task runners (docs) - prefect.futures source - Ray integration source If you share which runner you’re using in your outer flow (ThreadPool vs Ray vs PrefectTaskRunner), I can tailor the await/wait pattern exactly to that setup.

Short answer: No. In Prefect 3.x there’s no

Task.map(...)

, and

wait

is a function (not chainable off a list/future). Use

.submit(...)

in a loop/comprehension, then wait/resolve with Prefect’s futures utilities. Do this instead: - Sync flow (any runner)

from prefect import flow, task
from prefect.futures import resolve_futures_to_results

@task
def my_task(i: int) -> int:
    return i * 2

@flow
def main():
    futures = [my_task.submit(i) for i in range(10)]
    results = resolve_futures_to_results(futures)  # waits + unwraps
    return results

- If you need explicit waiting or streaming:

from prefect.futures import wait, as_completed

# Wait for all
done, not_done = wait(futures)
results = [f.result() for f in done]

# Or process as they complete
for f in as_completed(futures):
    print(f.result())

- Async flow with PrefectTaskRunner (only runner whose futures support async APIs):

import asyncio
from prefect import flow, task
from prefect.task_runners import PrefectTaskRunner

@task
def my_task(i: int) -> int:
    return i * 2

@flow(task_runner=PrefectTaskRunner())
async def main_async():
    futures = [my_task.submit(i) for i in range(10)]
    results = await asyncio.gather(*(f.result_async() for f in futures))
    return results

- Async flow with ThreadPool/ProcessPool/Ray:

from prefect.futures import resolve_futures_to_results

@flow  # e.g., task_runner=ThreadPoolTaskRunner() or RayTaskRunner()
async def main_async():
    futures = [my_task.submit(i) for i in range(10)]
    results = resolve_futures_to_results(futures)  # preferred
    # or: results = await asyncio.to_thread(resolve_futures_to_results, futures)
    return results

References: - Task runners (docs) - prefect.futures source

Totally—`wait` is available and useful. What it is -

from prefect.futures import wait

- It’s synchronous and returns a DoneAndNotDoneFutures namedtuple: - `done`: set of futures that finished before the timeout - `not_done`: set of futures still running after the timeout - Works with Prefect futures from ThreadPool/ProcessPool/Ray/Distributed/Flow-run futures - No exception on timeout; it just returns with some in

not_done

- There’s no module-level `wait_async`; only distributed/flow-run futures have per-future `wait_async()`/`result_async()` Typical usage

from prefect import flow, task
from prefect.futures import wait

@task
def work(i: int) -> int:
    import time; time.sleep(i)
    return i * 2

@flow
def main():
    futures = [work.submit(i) for i in [1, 2, 3]]

    done, not_done = wait(futures, timeout=5)

    results = []
    for f in done:
        try:
            results.append(f.result())  # or f.result(raise_on_failure=False)
        except Exception as e:
            # handle failures from done tasks
            ...

    if not_done:
        # still running or timed out
        for f in not_done:
            # you can inspect f.state or wait again later
            ...

    return results

Async flows - You can call

wait()

inside an async flow; it blocks briefly to resolve futures. - For PrefectTaskRunner/flow-run futures only, you can stay fully async:

# only for PrefectDistributedFuture / PrefectFlowRunFuture
  results = await asyncio.gather(*(f.result_async() for f in futures))

- For ThreadPool/ProcessPool/Ray futures in an async flow, prefer: -

resolve_futures_to_results(futures)

(simple), or -

await asyncio.to_thread(resolve_futures_to_results, futures)

to avoid blocking the event loop If you share which task runner(s) you’re using in the flow, I can suggest the exact waiting pattern to minimize blocking.

Absolutely—here’s your example adapted to use with_options so the RayTaskRunner is initialized at runtime. The Ray subflow is defined without a task runner, then configured inside the main flow right before it’s called.

import asyncio
from prefect import flow, task
from prefect.futures import resolve_futures_to_results, as_completed
from prefect.task_runners import ThreadPoolTaskRunner
from prefect_ray.task_runners import RayTaskRunner
# You can also import as: from prefect_ray import RayTaskRunner

# Cheap async task — keep on threads
@task
async def cheap_task(i: int) -> str:
    await asyncio.sleep(0.1)
    return f"cheap-{i}"

# Heavy sync task — run on Ray
@task
def heavy_task(i: int) -> int:
    return sum(j*j for j in range(i * 10000))  # CPU-y example

# Ray subflow: define without a task runner
@flow(name="ray-stage")
async def ray_stage(items: list[int]) -> list[int]:
    fs = [heavy_task.submit(i) for i in items]
    # Ray futures are not awaitable; use .result() or prefect.futures utilities
    return [f.result() for f in fs]  # or resolve_futures_to_results(fs)

# Main flow: async + threads for cheap work; configure Ray subflow at runtime
@flow(name="main", task_runner=ThreadPoolTaskRunner(max_workers=8))
async def main():
    # Fan out cheap async tasks on a thread pool
    cheap_fs = [cheap_task.submit(i) for i in range(10)]

    # Preferred: resolve all results (blocking call inside the async flow)
    cheap_results = resolve_futures_to_results(cheap_fs)

    # Or process incrementally as tasks complete
    for fut in as_completed(cheap_fs):
        _ = fut.result()  # do something incrementally

    # Configure Ray task runner at runtime (instantiate here)
    ray_stage_rt = ray_stage.with_options(
        task_runner=RayTaskRunner(
            # If connecting to a cluster, set address="<ray://host:port>"
            init_kwargs={"num_cpus": 4}
        )
    )

    # Hand heavy parts to Ray subflow (await the subflow itself)
    ray_results = await ray_stage_rt(items=list(range(25)))

    return {"cheap": cheap_results, "ray": ray_results}

if __name__ == "__main__":
    asyncio.run(main())

Notes: - You can also pass the class instead of an instance if you don’t need `init_kwargs`:

ray_stage.with_options(task_runner=RayTaskRunner)

. - To connect to an existing Ray cluster, use

RayTaskRunner(address="<ray://127.0.0.1:10001>", init_kwargs={...})

.