A couple of cleanups will make this pattern robust in Prefect 3.x: - You don’t need to branch on

f.state

inside `as_completed`; just call

f.result(raise_on_failure=True)

and let it raise on failures. - If you’re relying on tag-based concurrency, make sure each submitted task has the tag and that the concurrency limit exists for that tag. - Your

raise f.state.result(...)

will not do what you expect; call

f.result(raise_on_failure=True)

(or

f.exception()

) instead. Here’s a corrected, self-contained sketch:

from prefect import tags
from prefect.futures import as_completed

results = []
futures = []

# Optional: ensure these limits once, not per submit (idempotent if you keep it)
# for m in _models: ensure_tag_limit(m.name, 1)

for m in _models:
    ensure_tag_limit(m.name, 1)  # if this is idempotent, fine to keep here
    with tags(m.name):  # ensure the concurrency limit (by tag) will apply
        futures.append(
            load_dlt_table_task.submit(
                table=m,
                progress=progress,
                reset=str(reset).strip().lower() in ("true", "1", "yes", "y"),
                **pipeline_kwargs,
            )
        )

for f in as_completed(futures):
    # will raise immediately if the task failed
    results.append(f.result(raise_on_failure=True))

return results

CLI: create/inspect limits by tag - Create a limit:

prefect concurrency-limit create <tag> <limit>

- Inspect:

prefect concurrency-limit inspect <tag>

- List:

prefect concurrency-limit ls

Docs and references - Tag-based concurrency limits: How to: tag-based concurrency limits - Futures utilities: prefect.futures module If you prefer to fail fast and stop processing remaining futures after the first error, wrap the

f.result(...)

in a try/except and decide whether to cancel remaining work. Do you want fail-fast behavior or “collect everything, then raise” behavior?