https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I have quite a complex DAG for my fl...
# ask-marvin
s
@Marvin I have quite a complex DAG for my flow. So, is it okay for me to chain tasks so that I can reuse a particular task in more than one subsequent tasks, e.g.:
Copy code
@flow(task_runner=DaskTaskRunner(cluster_class=PBSCluster,  # type: ignore[arg-type]
                                 cluster_kwargs=pbs_cluster_kwargs,
                                 adapt_kwargs={"maximum": MAX_WORKERS}),
      log_prints=True,
)
def sfc_ref():
    ...
    # (aa2/aa3) 6s atmos correction
    # outputs: zdirectirr.img, zdiffuseirr.img, zsfcrad.img
    sixs_fs = atmos_correct.map(to_process, toa_rad)

    # Adjust diffuse irradiance for the sky view factor of each pixel.
    # (aa2/aa3) zdiffuseirradj.img
    zdiffuse_irradj = adjust_diff_irrad.map(to_process, sixs_fs)

    # direct irradiance, (aa2/aa3) zdirectirradj.img
    zdirect_irradj = compute_directirr.map(to_process, sixs_fs)

    # final sfc reflectance aba/abb
    aba_abbs = sfc_ref.map(
        sixs_fs, zdiffuse_irradj, zdirect_irradj, zincidence
    )

    done, not_done = wait(aba_abbs)
Is this possible? Can I map futures from a given task on to several subsequent tasks?
m
✅ thought for 151.5 seconds
Short answer: yes. In Prefect 3.x you can absolutely fan‑out a mapped task’s futures to multiple downstream mapped tasks. Passing a 
PrefectFutureList
(the result of 
.map(...)
) into other 
.map(...)
calls is a supported and common pattern; Prefect will build the dependency graph and schedule everything correctly. Your 
wait(aba_abbs)
at the end is fine too. A few clarifications and tips to make this work smoothly: - Mapping semantics are zip‑by‑position. If you pass multiple iterables to 
.map(...)
, they must be the same length. If you want to broadcast a constant across the mapping, wrap it with `unmapped()`:
Copy code
from prefect import unmapped

  zdiffuse_irradj = adjust_diff_irrad.map(to_process, sixs_fs)                  # zip to_process and sixs_fs
  zdirect_irradj  = compute_directirr.map(to_process, sixs_fs)                  # reuse sixs_fs again
  aba_abbs        = sfc_ref_task.map(sixs_fs, zdiffuse_irradj, zdirect_irradj, unmapped(zincidence))
- If 
to_process
is a constant (not per‑item), also use 
unmapped(to_process)
. - If 
to_process
is per‑item and aligns with 
sixs_fs
, your original zipping is correct. - Make sure you are calling 
.map
on a task, not a flow. In your snippet the flow is named 
sfc_ref
and you also call 
sfc_ref.map(...)
. If that’s the exact code, it will refer to the flow (flows don’t support 
.map
). Ensure the final reflectance is a task (e.g., 
@task def sfc_ref_task(...)
) and use 
sfc_ref_task.map(...)
, or rename the flow to avoid the name collision. - You can wait or collect results in a few ways: - 
done, not_done = wait(aba_abbs)
- 
results = aba_abbs.result()
# blocks and returns a list of results - DaskTaskRunner specifics: - Reusing the same upstream futures in multiple downstream mappings is efficient with Dask; the same underlying distributed futures are referenced and scheduled without recomputation. - Ensure everything passed between tasks is serializable by Dask workers. - If running as a script, include the 
if __name__ == "__main__":
guard when using Dask. If you share whether 
to_process
and 
zincidence
are per‑item or constants, I can suggest the exact `.map(...)`/`unmapped(...)` arrangement for your case.
s
@Marvin Using deployments yaml spedification, how should I specify a boolean for an environment variable:
Copy code
job_variables:
      env:
          IGNORE_CACHE: 'true'
m
✅ thought for 161.8 seconds
Short answer: quote it. In Prefect 3.x, all env vars are passed to the worker as strings. If you use an unquoted YAML boolean (`true`/`false`), YAML parses it as a bool and Prefect will coerce it to the string `"True"`/`"False"` (capitalized), which can be surprising. Use explicit strings in your `prefect.yaml`:
Copy code
deployments:
  - name: my-deployment
    entrypoint: flow.py:my_flow
    work_pool:
      name: my-pool
      job_variables:
        env:
          IGNORE_CACHE: "true"   # recommended
          DEBUG: "false"
In your flow code, parse it explicitly:
Copy code
import os
ignore_cache = os.getenv("IGNORE_CACHE", "false").lower() == "true"
Notes: - Unquoted YAML booleans become `"True"`/`"False"` after coercion; quoting avoids case surprises. - To unset an env var, set it to 
null
or omit it. Docs: Deployments (see job variables/env)
s
@Marvin How to temporarily set the logger level in a task?
m
✅ thought for 152.8 seconds
In Prefect 3.x, 
get_run_logger()
returns a PrefectLogAdapter that wraps a standard logger. You can temporarily raise/lower its level, then restore it. Quick inline pattern:
Copy code
from prefect import task
from prefect.logging import get_run_logger
import logging

@task
def my_task():
    logger = get_run_logger()
    underlying = logger.logger if hasattr(logger, "logger") else logger

    old = underlying.level
    try:
        underlying.setLevel(logging.DEBUG)
        logger.debug("This will be emitted during the temporary DEBUG window")
    finally:
        underlying.setLevel(old)

    logger.debug("Back to previous level (likely not emitted)")
Reusable context manager:
Copy code
from contextlib import contextmanager
import logging

@contextmanager
def temporary_log_level(logger, level):
    underlying = logger.logger if hasattr(logger, "logger") else logger
    original = underlying.level
    try:
        underlying.setLevel(level)
        yield
    finally:
        underlying.setLevel(original)
Usage in a task:
Copy code
from prefect import task
from prefect.logging import get_run_logger
import logging

@task
def my_task():
    logger = get_run_logger()
    with temporary_log_level(logger, logging.DEBUG):
        logger.debug("Temporarily visible")
Notes: - Handler levels still apply after the logger level. Prefect’s default API handler accepts all levels, so adjusting the logger level is usually sufficient. - Alternative: 
LogEavesdropper
can temporarily set level on a named logger and restore it after the block, while capturing output:
Copy code
from prefect.logging import LogEavesdropper
import logging

with LogEavesdropper("prefect.task_runs", level=logging.DEBUG):
    logger = get_run_logger()
    logger.debug("Captured at DEBUG")
s
@Marvin So, coulf I use 
LogEavesdropper
context manager inside a task as well?
Copy code
from prefect import task
from prefect.logging import get_run_logger


@task
def my_task():
    with LogEavesdropper("prefect.task_runs", level=logging.DEBUG):
        logger = get_run_logger()
        logger.debug("Captured at DEBUG")
Will it log DEBUG level messages coming/emitted by third-party libraries within the task?
m
✅ thought for 134.3 seconds
Yes—you can use LogEavesdropper inside a task, and it will work as you expect. - It attaches a temporary handler to the named logger you specify, raises that logger’s level for the duration of the with-block, captures messages that meet or exceed that level, then restores the original level on exit. - It captures logs from third-party libraries too, as long as you eavesdrop on the right logger name and those logs propagate to the target logger. Examples: - Capture Prefect task-run logs at DEBUG during the block:
Copy code
from prefect import task
from prefect.logging import get_run_logger, LogEavesdropper
import logging

@task
def my_task():
    with LogEavesdropper("prefect.task_runs", level=logging.DEBUG) as e:
        logger = get_run_logger()
        logger.debug("Captured at DEBUG")
    print(e.text())
- Capture third-party library logs (e.g., requests/urllib3):
Copy code
from prefect import task
from prefect.logging import get_run_logger, LogEavesdropper
import logging

@task
def my_task():
    with LogEavesdropper("urllib3", level=logging.DEBUG) as e:
        import urllib3
        urllib3_logger = logging.getLogger("urllib3")
        urllib3_logger.debug("TLS handshake details...")
    print(e.text())
- Capture “everything” by targeting the root logger:
Copy code
with LogEavesdropper("", level=logging.DEBUG) as e:
    # any logger that propagates to root will be captured
    pass
Caveats: - Propagation matters: child loggers are captured only if they have propagate=True (default). If a library sets propagate=False on its logger, those records won’t reach your target logger and won’t be captured. - The level you pass is applied during the block; after exit, the original level is restored. - The default formatter is “[LEVEL]: message”. You can set a custom formatter with 
e.setFormatter(logging.Formatter(...))
. API reference: LogEavesdropper
s
@Marvin I have the following ending of my flow, where the last futures just return None:
Copy code
@flow()
def my_flow() -> Path:  
    # final sfc reflectance aba/abb paths
    aba_abbs: PrefectFutureList[str] = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )
    # save them, these futures return None
    fs: PrefectFutureList = save_qvf_file.map(aba_abbs)

    done, not_done = wait(fs)

    # collect aba/abbs completed
    out_fns: list[str] = []
    for f in done:
        if f.state.is_completed():
            out_fns.append(aba_abbs[i].result())
Copy code
What is the best way to get the result of my `aba_abbs`?
m
✅ thought for 12.1 seconds
You usually don’t want to use 
wait(fs)
and then manually index back into 
aba_abbs
. If your goal is: “wait until the saves are done, then collect the successful 
aba_abbs
results”, the simplest pattern is:
Copy code
python
from prefect import flow

@flow
def my_flow() -> list[str]:
    aba_abbs = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )

    fs = save_qvf_file.map(aba_abbs)

    # Ensure the save tasks have finished
    fs.wait()

    # Then resolve the upstream mapped results
    out_fns = aba_abbs.result()

    return out_fns
A few notes: * 
aba_abbs
is already a list of futures for the values you want. * 
fs
depends on 
aba_abbs
, so once the save tasks finish, the corresponding 
aba_abbs
tasks must have finished too. * 
aba_abbs.result()
is the cleanest way to retrieve all mapped results. If you only want the ones whose save succeeded, then iterate over both together:
Copy code
python
from prefect import flow

@flow
def my_flow() -> list[str]:
    aba_abbs = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )
    fs = save_qvf_file.map(aba_abbs)

    fs.wait()

    out_fns = []
    for aba_future, save_future in zip(aba_abbs, fs):
        if save_future.state.is_completed():
            out_fns.append(aba_future.result())

    return out_fns
Your current snippet also has an issue: 
i
is not defined in the loop over 
done
, so 
aba_abbs[i]
won’t work as written. If you want, I can also show the Prefect 3-native pattern for mapped task collection with better failure handling.
s
@Marvin Sure.
m
✅ thought for 80.9 seconds
Yep — in Prefect 3.x, the most natural pattern is: * map the upstream task * map the downstream side-effect task that returns 
None
* wait on the downstream futures if needed * then call 
.result()
on the upstream 
PrefectFutureList
For your case, I’d write it like this:
Copy code
python
from pathlib import Path
from prefect import flow

@flow
def my_flow() -> list[str]:
    aba_abbs = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )

    # side-effect task, returns None
    save_futures = save_qvf_file.map(aba_abbs)

    # wait until all saves finish
    save_futures.wait()

    # collect the upstream mapped results in order
    out_fns = aba_abbs.result()

    return out_fns
A few important notes: * 
aba_abbs.result()
returns the mapped results in order * because 
save_qvf_file.map(aba_abbs)
depends on 
aba_abbs
, once 
save_futures.wait()
finishes, the corresponding 
aba_abbs
tasks are already finished too * if all you need is “save completed, now give me the filenames”, this is the cleanest approach If you want partial failure handling — e.g. only return 
aba_abbs
whose corresponding save succeeded — then zip the futures together:
Copy code
python
from prefect import flow

@flow
def my_flow() -> list[str]:
    aba_abbs = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )
    save_futures = save_qvf_file.map(aba_abbs)

    save_futures.wait()

    out_fns = []
    for aba_future, save_future in zip(aba_abbs, save_futures):
        if save_future.state.is_completed():
            out_fns.append(aba_future.result())

    return out_fns
If you also want to avoid raising when some upstream/downstream tasks fail, you can inspect states explicitly:
Copy code
python
from prefect import flow

@flow
def my_flow() -> list[str]:
    aba_abbs = std_ref.map(
        to_process, zdiffuse_irradj, zdirect_irradj, out_dir=destdir
    )
    save_futures = save_qvf_file.map(aba_abbs)

    save_futures.wait()

    out_fns = []
    for aba_future, save_future in zip(aba_abbs, save_futures):
        if save_future.state.is_completed() and aba_future.state.is_completed():
            out_fns.append(aba_future.result())

    return out_fns
Your original code has two issues: * 
i
is undefined in 
aba_abbs[i].result()
* 
wait(fs)
returns sets of futures, so you lose the original mapped ordering/index unless you track it yourself So the short answer is: Best practice: use 
save_futures.wait()
and then 
aba_abbs.result()
If you need only successful pairs: iterate with 
zip(aba_abbs, save_futures)
If you want, I can also show a version using 
submit
instead of 
map
, or one that returns 
Path
objects cleanly.
s
@Marvin Show me one that returns 
Path
objects cleanly.
m
✅ thought for 13.0 seconds
Sure — the cleanest pattern is to have the upstream mapped task return 
Path
directly, and let the save task accept a 
Path
and return 
None
. Example:
Copy code
python
from pathlib import Path
from prefect import flow, task

@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.qvf"

    # create/write whatever needs to be produced
    # ...

    return out_path

@task
def save_qvf_file(path: Path) -> None:
    # persist / upload / finalize side effect
    # ...
    return None

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    save_futures = save_qvf_file.map(aba_abbs)

    # wait for all save side effects to finish
    save_futures.wait()

    # collect upstream Path results in order
    out_paths = aba_abbs.result()

    return out_paths
If you only want paths whose save step succeeded:
Copy code
python
from pathlib import Path
from prefect import flow, task

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    save_futures = save_qvf_file.map(aba_abbs)

    save_futures.wait()

    out_paths: list[Path] = []
    for path_future, save_future in zip(aba_abbs, save_futures):
        if save_future.state.is_completed() and path_future.state.is_completed():
            out_paths.append(path_future.result())

    return out_paths
A nice refinement is to make 
save_qvf_file
return the same 
Path
after saving, so you only collect one mapped result list:
Copy code
python
from pathlib import Path
from prefect import flow, task

@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.qvf"
    # generate file
    return out_path

@task
def save_qvf_file(path: Path) -> Path:
    # save/finalize/upload path
    return path

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    saved_paths = save_qvf_file.map(aba_abbs)

    return saved_paths.result()
That last version is often the most ergonomic because the final futures already represent “successfully saved `Path`s”. If you want, I can show which of these three patterns I’d recommend for production and why.
✅ thought for 8.7 seconds
Yes — that pattern is allowed in Prefect 3.x, and it’s a good one. Your flow is effectively saying: 1. map 
std_ref
to produce `Path`s 2. map 
save_qvf_file
over those `Path`s 3. return the resolved results from the second mapped task That works fine as long as 
save_qvf_file
really returns the 
Path
after the save/finalize step succeeds.
Copy code
python
from pathlib import Path
from prefect import flow, task

@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.qvf"
    # generate file
    return out_path

@task
def save_qvf_file(path: Path) -> Path:
    # save/finalize/upload path
    return path

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    saved_paths = save_qvf_file.map(aba_abbs)

    return saved_paths.result()
Why this is nice: * 
saved_paths.result()
gives you 
list[Path]
* the returned values represent successfully completed save steps * you don’t need a separate 
wait()
, because 
.result()
already waits for completion * it keeps the dataflow simple and ordered A couple caveats: * if any mapped 
save_qvf_file
task fails, 
saved_paths.result()
will raise by default * if 
save_qvf_file
just echoes the same path back, that’s totally fine — it’s a common pattern for side-effect tasks when you want the downstream result to stay useful So yes: this is valid, idiomatic, and probably the cleanest version for your case. If you want partial-failure behavior, then don’t call 
saved_paths.result()
directly; inspect each future instead.
s
@Marvin What about this pattern, is this allowed?
Copy code
@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map()
   
    aba_abbs = std_ref.map()

    saved_paths = save_qvf_file.map(aba_abbs + other_products)

    return saved_paths.result()
m
✅ thought for 11.9 seconds
Not as written. The main issue is this part:
Copy code
python
saved_paths = save_qvf_file.map(aba_abbs + other_products)
Whether this is allowed depends on what 
aba_abbs
and 
other_products
actually are, but generally you should not rely on concatenating `PrefectFutureList`s and passing that directly to 
.map()
. Why: * 
.map()
expects an iterable of inputs to map over * 
aba_abbs
and 
other_products
are `PrefectFutureList`s * 
aba_abbs + other_products
will produce a plain Python 
list
, not necessarily a 
PrefectFutureList
* while Prefect can often resolve futures inside collections, combining mapped outputs this way is not the clearest or safest pattern Also, your snippet has another issue:
Copy code
python
aba_abbs = std_ref.map()
That reassigns 
aba_abbs
and discards the earlier mapped result, so I assume that line is accidental. If your goal is: map two independent producers, combine all produced paths, then run one mapped save over the combined paths, the safer pattern is to resolve first, then remap:
Copy code
python
from pathlib import Path
from prefect import flow

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(...)

    all_paths: list[Path] = aba_abbs.result() + other_products.result()

    saved_paths = save_qvf_file.map(all_paths)

    return saved_paths.result()
That is definitely valid. If you want to preserve parallelism as much as possible, you can also wait for both groups together, then combine:
Copy code
python
from pathlib import Path
from prefect import flow

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(...)

    all_paths = aba_abbs.result() + other_products.result()

    saved_paths = save_qvf_file.map(all_paths)

    return saved_paths.result()
If instead you want 
save_qvf_file
to start saving each branch as soon as that branch is ready, then map them separately:
Copy code
python
from pathlib import Path
from prefect import flow

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(...)

    saved_aba_abbs = save_qvf_file.map(aba_abbs)
    saved_other = save_qvf_file.map(other_products)

    return saved_aba_abbs.result() + saved_other.result()
That last version is often better because: * each branch can flow independently * no need to manually combine futures before mapping * clearer dependency graph So short answer: * 
save_qvf_file.map(aba_abbs + other_products)
— I would not recommend this pattern * 
save_qvf_file.map(aba_abbs.result() + other_products.result())
— yes, allowed * 
save_qvf_file.map(...)
separately for each mapped branch — usually the best pattern If you want, I can verify from Prefect’s actual mapping semantics whether a raw concatenated list of futures is technically accepted, versus just being a bad style pattern.
s
@Marvin
Copy code
If you want, I can verify from Prefect's actual mapping semantics whether a raw concatenated list of futures is technically accepted, versus just being a bad style pattern.
Sure.
m
✅ thought for 130.2 seconds
Yes — technically, this is allowed in Prefect 3.x. So this pattern can work:
Copy code
python
combined = aba_abbs + other_products
saved_paths = save_qvf_file.map(combined)
Even though 
aba_abbs + other_products
gives you a plain Python 
list
rather than a 
PrefectFutureList
, Prefect can still map over it and resolve each 
PrefectFuture
element correctly. What’s happening: * 
aba_abbs
and 
other_products
are `PrefectFutureList`s * concatenating them produces a regular 
list
* 
Task.map(...)
only needs an iterable to map over * when each mapped 
save_qvf_file
task actually runs, Prefect resolves the future passed as that element into its final result So Prefect does not require a 
PrefectFutureList
specifically as input to 
.map()
. That means this is valid:
Copy code
python
from pathlib import Path
from prefect import flow

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(...)

    saved_paths = save_qvf_file.map(aba_abbs + other_products)

    return saved_paths.result()
A few practical notes: * 
saved_paths.result()
will wait for all save tasks and return 
list[Path]
* order will be preserved as: * all 
aba_abbs
first * then all 
other_products
* if any upstream future or any 
save_qvf_file
task fails, 
.result()
will raise by default So my updated recommendation is: * Allowed? Yes * Technically sound? Yes * Reasonable to use? Also yes, if combining two mapped branches into one downstream mapped task is what you want I’d still prefer making the combination explicit for readability:
Copy code
python
all_products = aba_abbs + other_products
saved_paths = save_qvf_file.map(all_products)
That’s a bit clearer than doing the concatenation inline. Also, one correction to your earlier example: this line would still be suspicious unless intentional:
Copy code
python
aba_abbs = std_ref.map()
because it overwrites the original 
aba_abbs
. For reference, I verified this against Prefect 3.x mapping semantics and `Task.map`’s signature: * 
Task.map(..., wait_for=None, deferred=False, **kwargs)
returns a 
PrefectFutureList
* Prefect resolves nested futures elementwise at task execution time, so a raw list containing futures is acceptable input If you want, I can show a polished final version of your multi-branch flow using this pattern.
s
@Marvin Sure
m
✅ thought for 9.6 seconds
Here’s a clean version of the multi-branch pattern using combined mapped outputs and returning `list[Path]`:
Copy code
python
from pathlib import Path
from prefect import flow, task

@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.qvf"
    # generate product
    return out_path

@task
def other_prod(item, out_dir: Path) -> Path:
    out_path = out_dir / f"{item}.other"
    # generate other product
    return out_path

@task
def save_qvf_file(path: Path) -> Path:
    # save / finalize / upload
    return path

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    all_products = aba_abbs + other_products

    saved_paths = save_qvf_file.map(all_products)

    return saved_paths.result()
Why this is good: * both producer branches run independently * 
all_products
can be a plain Python list of futures * 
save_qvf_file.map(all_products)
is valid in Prefect 3.x * returning the path from 
save_qvf_file
makes the final collection straightforward If you want a slightly more defensive version for partial failures, use separate save branches instead:
Copy code
python
from pathlib import Path
from prefect import flow, task

@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    saved_aba_abbs = save_qvf_file.map(aba_abbs)
    saved_other = save_qvf_file.map(other_products)

    out_paths: list[Path] = []

    for f in saved_aba_abbs:
        if f.state.is_completed():
            out_paths.append(f.result())

    for f in saved_other:
        if f.state.is_completed():
            out_paths.append(f.result())

    return out_paths
Use the first version if: * you want one combined downstream save stage * you’re okay with the whole 
.result()
raising if any mapped save fails Use the second if: * you want easier branch-by-branch failure handling No specific docs link came from the code-verification tools here, but the relevant verified APIs are: * 
Task.map(... ) -> PrefectFutureList
* 
PrefectFutureList.result()
* 
prefect.futures.wait
/ 
as_completed
If you want, I can also show the same flow written to preserve type clarity with intermediate annotations like 
PrefectFutureList[Path]
.
s
@Marvin Sure
m
✅ thought for 11.8 seconds
Absolutely — here’s the same pattern with explicit type annotations for readability.
Copy code
python
from pathlib import Path

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


@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.qvf"
    # generate product
    return out_path


@task
def other_prod(
    item,
    out_dir: Path,
) -> Path:
    out_path = out_dir / f"{item}.other"
    # generate other product
    return out_path


@task
def save_qvf_file(path: Path) -> Path:
    # save / finalize / upload
    return path


@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs: PrefectFutureList[Path] = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    other_products: PrefectFutureList[Path] = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    all_products: list = aba_abbs + other_products

    saved_paths: PrefectFutureList[Path] = save_qvf_file.map(all_products)

    return saved_paths.result()
A couple small notes on typing: * 
aba_abbs
and 
other_products
are nicely typed as 
PrefectFutureList[Path]
* 
aba_abbs + other_products
becomes a plain Python 
list
, not a 
PrefectFutureList
* if you want to be more explicit, you could annotate that as:
Copy code
python
all_products: list = aba_abbs + other_products
or, conceptually:
Copy code
python
all_products: list[object] = aba_abbs + other_products
But Python typing gets a little awkward here because that concatenated list actually contains futures whose eventual result is 
Path
, not 
Path
objects yet. So in practice, I’d probably write:
Copy code
python
all_products = aba_abbs + other_products
and let inference handle it. If you want maximum clarity without the awkward intermediate type, this is probably the nicest balance:
Copy code
python
from pathlib import Path

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


@task
def std_ref(
    item,
    zdiffuse_irradj,
    zdirect_irradj,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.qvf"


@task
def other_prod(item, out_dir: Path) -> Path:
    return out_dir / f"{item}.other"


@task
def save_qvf_file(path: Path) -> Path:
    return path


@flow
def my_flow(destdir: Path) -> list[Path]:
    aba_abbs: PrefectFutureList[Path] = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products: PrefectFutureList[Path] = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    saved_paths: PrefectFutureList[Path] = save_qvf_file.map(
        aba_abbs + other_products
    )

    return saved_paths.result()
That keeps the important types explicit and avoids over-typing the combined list. One small caveat: some static type checkers may not love importing and parameterizing 
PrefectFutureList
depending on your Prefect version/type stubs, even though it is valid at runtime. If you want, I can give you a mypy-friendly version next.
s
@Marvin give me the 
mypy-friendly
version
m
✅ thought for 10.9 seconds
Here’s a more mypy-friendly version that avoids annotating the concatenated futures list directly.
Copy code
python
from pathlib import Path
from typing import Iterable

from prefect import flow, task


@task
def std_ref(
    item: str,
    zdiffuse_irradj: float,
    zdirect_irradj: float,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.qvf"


@task
def other_prod(
    item: str,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.other"


@task
def save_qvf_file(path: Path) -> Path:
    # save / finalize / upload
    return path


@flow
def my_flow(
    destdir: Path,
    to_process: list[str],
    other_to_process: list[str],
    zdiffuse_irradj: list[float],
    zdirect_irradj: list[float],
) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )

    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    combined_futures = [*aba_abbs, *other_products]

    saved_paths = save_qvf_file.map(combined_futures)

    return saved_paths.result()
Why this is more mypy-friendly: * avoids importing and parameterizing 
PrefectFutureList
* avoids trying to precisely annotate the type of 
aba_abbs + other_products
* 
combined_futures = [*aba_abbs, *other_products]
is often easier for type checkers than 
aba_abbs + other_products
* the flow return type stays clean: 
list[Path]
If you want to be even more conservative for static typing, you can avoid combining futures altogether and combine resolved results instead:
Copy code
python
from pathlib import Path

from prefect import flow, task


@task
def std_ref(
    item: str,
    zdiffuse_irradj: float,
    zdirect_irradj: float,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.qvf"


@task
def other_prod(
    item: str,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.other"


@task
def save_qvf_file(path: Path) -> Path:
    return path


@flow
def my_flow(
    destdir: Path,
    to_process: list[str],
    other_to_process: list[str],
    zdiffuse_irradj: list[float],
    zdirect_irradj: list[float],
) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    all_paths: list[Path] = aba_abbs.result() + other_products.result()

    saved_paths = save_qvf_file.map(all_paths)

    return saved_paths.result()
That version is usually the easiest for mypy to understand, though it changes execution behavior slightly because the combined save stage won’t start until both producer groups have resolved. So the tradeoff is: * best runtime parallelism: combine futures * best mypy simplicity: combine resolved 
list[Path]
If you want, I can also give you a version that is both mypy-clean and preserves branch-level parallelism by saving each branch separately.
s
@Marvin Sure
m
✅ thought for 11.7 seconds
Yep — this is the version I’d use if you want both: * clean typing for mypy * good parallelism * clear failure boundaries per branch
Copy code
python
from pathlib import Path

from prefect import flow, task


@task
def std_ref(
    item: str,
    zdiffuse_irradj: float,
    zdirect_irradj: float,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.qvf"


@task
def other_prod(
    item: str,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.other"


@task
def save_qvf_file(path: Path) -> Path:
    # save / finalize / upload
    return path


@flow
def my_flow(
    destdir: Path,
    to_process: list[str],
    other_to_process: list[str],
    zdiffuse_irradj: list[float],
    zdirect_irradj: list[float],
) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    saved_aba_abbs = save_qvf_file.map(aba_abbs)
    saved_other_products = save_qvf_file.map(other_products)

    return saved_aba_abbs.result() + saved_other_products.result()
Why this is a nice compromise: * mypy only has to understand: * mapped tasks returning paths * 
.result()
returning 
list[Path]
* each branch can start saving as soon as its own upstream products are ready * you avoid the awkward “list of futures from concatenation” typing problem * the final return type is naturally 
list[Path]
Behavior-wise: * 
std_ref
and 
other_prod
can run independently * 
save_qvf_file.map(aba_abbs)
can start as soon as 
aba_abbs
items are ready * same for 
other_products
* the final concatenation happens only after both save branches are complete If you want partial-failure handling without blowing up on the first failed save, use explicit future inspection instead of `.result()`:
Copy code
python
from pathlib import Path

from prefect import flow, task


@task
def std_ref(
    item: str,
    zdiffuse_irradj: float,
    zdirect_irradj: float,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.qvf"


@task
def other_prod(
    item: str,
    out_dir: Path,
) -> Path:
    return out_dir / f"{item}.other"


@task
def save_qvf_file(path: Path) -> Path:
    return path


@flow
def my_flow(
    destdir: Path,
    to_process: list[str],
    other_to_process: list[str],
    zdiffuse_irradj: list[float],
    zdirect_irradj: list[float],
) -> list[Path]:
    aba_abbs = std_ref.map(
        to_process,
        zdiffuse_irradj,
        zdirect_irradj,
        out_dir=destdir,
    )
    other_products = other_prod.map(
        other_to_process,
        out_dir=destdir,
    )

    saved_aba_abbs = save_qvf_file.map(aba_abbs)
    saved_other_products = save_qvf_file.map(other_products)

    saved_aba_abbs.wait()
    saved_other_products.wait()

    out_paths: list[Path] = []

    for f in saved_aba_abbs:
        if f.state.is_completed():
            out_paths.append(f.result())

    for f in saved_other_products:
        if f.state.is_completed():
            out_paths.append(f.result())

    return out_paths
So if you want my blunt recommendation: For mypy + readability + runtime behavior, I’d prefer separate save maps per branch over combining future lists.
Open in Slack
PreviousNext
Powered by