Great question — this is a classic fan-out/fan-in pattern in Prefect 3.x. Here are the recommended approaches, depending on whether you want to start processing as soon as each search finishes (streaming) or wait until all searches are done (batch). Assumptions: Prefect 3.x; your API calls are I/O bound. Core building blocks - Use task mapping for fan-out:

search_api.map(queries)

and

process_record.map(records)

. - Use

as_completed

to start post-processing as soon as each search returns. - Control concurrency with a task runner (threads for I/O) and optionally global concurrency limits/rate limits for APIs. - Add retries with backoff on the API calls. Pattern A — Streamed: begin processing each list as soon as it’s ready This minimizes end-to-end latency and doesn’t wait for all searches to finish.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.futures import as_completed
from prefect.tasks import exponential_backoff
from prefect.concurrency.sync import rate_limit

@task(retries=3, retry_delay_seconds=exponential_backoff(2), log_prints=True)
def search_api(query: str) -> list[dict]:
    # Optional: enforce API rate limits globally (see CLI below)
    with rate_limit("my-api", timeout_seconds=30, strict=False):
        # call external API here
        # return a list of records
        ...

@task
def process_record(rec: dict) -> dict:
    # post-process a single record
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def pipeline(queries: list[str]):
    # Fan-out searches concurrently
    search_futures = search_api.map(queries)

    processed = []
    # As each search finishes, map the processing over its returned list
    for f in as_completed(search_futures):
        records = f.result()   # this is the list from that search
        processed.extend(process_record.map(records))

    # Optionally wait for all processing to complete and return results
    return [p.result() for p in processed]

Pattern B — Batch: wait for all searches, then process all records This is simpler but waits for the entire first stage to finish.

from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.tasks import exponential_backoff
from prefect.concurrency.sync import rate_limit

@task(retries=3, retry_delay_seconds=exponential_backoff(2))
def search_api(query: str) -> list[dict]:
    with rate_limit("my-api", timeout_seconds=30, strict=False):
        ...

@task
def process_record(rec: dict) -> dict:
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def pipeline(queries: list[str]):
    lists = search_api.map(queries).result()  # resolve to list-of-lists
    flat_records = [rec for lst in lists for rec in lst]
    processed = process_record.map(flat_records)
    return [p.result() for p in processed]

Concurrency and rate limiting - Task runner: For I/O-bound HTTP calls, threads are ideal. Control concurrency with

max_workers

on

ThreadPoolTaskRunner

. For CPU-heavy post-processing, consider

ProcessPoolTaskRunner

or Dask if you need to scale out. - Global Concurrency Limits and Rate Limits: Use these to respect third-party API limits across all flows. - Create a limit: - List commands:

prefect gcl --help

- Create:

prefect gcl create my-api --limit 10 --slot-decay-per-second 10

- Use in tasks:

from prefect.concurrency.sync import rate_limit, concurrency

    # Rate limit (token bucket style)
    with rate_limit("my-api", timeout_seconds=30, strict=False):
        ...

    # Or a simple slot-based concurrency cap
    with concurrency("my-api-requests", occupy=1, timeout_seconds=30, strict=False):
        ...

- CLI help verified: - Global concurrency limits docs -

prefect gcl create --help

confirmed via CLI Reliability tweaks - Retries with backoff on network calls:

-

@task(retries=3, retry_delay_seconds=exponential_backoff(2))

- Docs: Retries and backoff - If individual records can fail but you want the flow to continue, you can wrap them with

allow_failure

when mapping. - If you need to observe futures as they complete:

as_completed

is great;

wait

is useful to partition into done/not-done if you need timeouts. When to pick which pattern - Use Pattern A (streamed) if you want to minimize latency and start processing as soon as possible. - Use Pattern B (batch) if simplicity is more important and waiting for all searches first is okay. Helpful docs - Run work concurrently: https://docs-3.prefect.io/v3/how-to-guides/workflows/run-work-concurrently - Futures API: https://docs-3.prefect.io/v3/api-ref/python/prefect-futures - Dask integration (if you need distributed): https://docs-3.prefect.io/integrations/prefect-dask/index A couple clarifying questions to recommend the exact pattern: - Do you want processing to start as soon as each search finishes, or is batching fine? - Roughly how many queries and records are we talking about (to size concurrency and decide if chunking is needed)? - Any specific API rate limits we should enforce (e.g., X requests per second)?

Happy to! Here’s a practical deep-dive on

prefect.futures.as_completed

in Prefect 3.x. What it does - It’s a generator that yields task futures in the order they finish, not the order you submitted them. - Use it to start processing results as soon as they’re ready (great for streaming fan-out/fan-in). Key behavior - Order: Completion order, not submission order. - Timeouts: If you pass

timeout=...

, and no additional futures complete within that window, it raises

TimeoutError

. Futures that completed before the timeout are already yielded. - Failures: It yields futures regardless of state (Completed, Failed, Cancelled). Calling

future.result()

on a failed future raises unless you pass

raise_on_failure=False

. - Multiple uses: You can call

as_completed

on the same set of futures multiple times; each call creates a new generator. Completed futures will be yielded immediately. - Duplicates: If the same future appears more than once in the input list, it’s deduplicated internally. - Early exit: You can break out of the loop anytime. It doesn’t affect the running tasks. Signature

from prefect.futures import as_completed

for f in as_completed(futures, timeout=None):
    ...

Common patterns 1) Stream results as they finish (handle success/failure)

from prefect.futures import as_completed

futures = search_api.map(queries)  # returns a PrefectFutureList

for fut in as_completed(futures):
    try:
        result = fut.result()  # may raise if task failed
    except Exception as exc:
        # Handle a failed or cancelled task
        print(f"Search failed: {exc}")
        continue

    # Process the successful result
    handle(result)

2) With timeout and partial progress

from prefect.futures import as_completed
from concurrent.futures import TimeoutError  # Python’s TimeoutError

processed = 0

try:
    for fut in as_completed(futures, timeout=10):
        value = fut.result()
        handle(value)
        processed += 1
except TimeoutError:
    print(f"Timed out after processing {processed} results.")

3) Streaming a second mapping based on each completed future This is how you start processing each returned list as soon as it’s ready:

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

@task
def search_api(q) -> list[dict]:
    ...

@task
def process_record(rec: dict):
    ...

@flow
def pipeline(queries: list[str]):
    search_futures = search_api.map(queries)

    processed_futures = []
    for fut in as_completed(search_futures):
        records = fut.result()  # the list from that specific search
        processed_futures.extend(process_record.map(records))

    # Optionally wait for all processing to finish
    return [pf.result() for pf in processed_futures]

When to choose as_completed vs wait() - Use

as_completed

when you want a streaming pipeline (react to work finishing). - Use

wait(futures)

when you want a snapshot: split into done/not_done at a moment in time, then decide what to do next. Docs and references - API reference (futures): prefect.futures - Concurrency patterns: Run work concurrently If you share: - about how many futures (tasks) you expect, - whether you want to stop on first error or keep going, - and if you need a global timeout for the whole stage, I can tailor an example that matches your exact use case.

Streaming fan-out/fan-in is a workflow pattern where you: - Fan-out: launch many independent tasks in parallel (e.g., many API searches). - Streamed fan-in: as each upstream task finishes, immediately begin its downstream work (e.g., process each returned list), instead of waiting for all upstream tasks to finish. Why it’s useful - Lower end-to-end latency: you start processing early results right away. - Better throughput/backpressure: you don’t build a huge intermediate buffer in memory. - Resilient to long tails: slow searches don’t block processing of fast ones. In Prefect 3.x, you implement this with `as_completed`:

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ThreadPoolTaskRunner

@task
def search_api(query: str) -> list[dict]:
    # call external API
    ...

@task
def process_record(rec: dict) -> dict:
    # post-process one record
    ...

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def pipeline(queries: list[str]):
    # Fan-out: run all searches concurrently
    search_futs = search_api.map(queries)

    processed_futs = []

    # Streaming fan-in: as each search completes, immediately process its records
    for fut in as_completed(search_futs):
        records = fut.result()  # this search’s list
        processed_futs.extend(process_record.map(records))

    # Optionally wait for all processing to finish (true fan-in)
    return [pf.result() for pf in processed_futs]

Notes -

as_completed

yields futures in completion order, not submission order. - You can combine this with rate limits and retries on the API task. - If you want to persist or aggregate results incrementally (to a DB, S3, etc.), call that in the loop so slow searches don’t block fast ones. Good references - Concurrent execution patterns: Run work concurrently - Futures API: prefect.futures If you share your expected number of searches/records and any API rate limits, I can sketch a tuned version with proper concurrency caps and backoff.

Great question. In Prefect 3.x, the most common pattern is to keep record-level processing parallelized, then “reduce” all results into a single batch write task. That gives you the best of both worlds: fast parallel processing and one atomic DB write. Recommended strategies (pick based on batch size and failure semantics): - Map → Reduce to one batch write (simple, one transaction) - Map a pure “process_record” task over all records - Pass the mapped futures directly to a single downstream “write_batch” task - Prefect will resolve those futures into a list for you - Chunked processing → Single batch write (memory-friendly) - Chunk input first, map a “process_chunk” task, then send the list-of-lists to one “write_batch” task and flatten there - Chunked processing → Chunked writes (for very large batches or tight timeouts) - Process and write by chunk in separate tasks; use DB transactions per chunk and accept partial commits Example 1 — Map then reduce to a single batch write

from prefect import flow, task, unmapped, tags
from prefect.tasks import exponential_backoff

@task
def process_record(rec: dict) -> dict:
    # pure transform, no DB IO
    # ... do CPU/light IO work ...
    return {"id": rec["id"], "value": rec["value"] * 2}

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(2),
)
def write_batch(processed: list[dict]):
    # Use your DB client (SQLAlchemy example shown)
    from sqlalchemy import create_engine, text
    engine = create_engine("<postgresql+psycopg2://user:pass@host:5432/db>")

    # single transaction for the whole batch write
    with engine.begin() as conn:
        # Example bulk upsert — adapt to your dialect
        stmt = text("""
            INSERT INTO my_table (id, value)
            VALUES (:id, :value)
            ON CONFLICT (id) DO UPDATE SET value = EXCLUDED.value
        """)
        conn.execute(stmt, processed)

@flow
def process_and_write(records: list[dict]):
    processed_futures = process_record.map(records)

    # Optional: limit concurrency on the write using a tag
    with tags("db_write"):
        write_batch(processed_futures)

# process_and_write(records)  # call with your batch

Example 2 — Chunk first to control memory, then single batch write

from prefect import flow, task, tags
from prefect.utilities.collections import batched_iterable
from prefect.tasks import exponential_backoff

@task
def process_chunk(chunk: list[dict]) -> list[dict]:
    return [{"id": r["id"], "value": r["value"] * 2} for r in chunk]

@task(
    retries=3,
    retry_delay_seconds=exponential_backoff(2),
)
def write_batch(all_processed: list[list[dict]]):
    # flatten
    flat = [item for sub in all_processed for item in sub]
    # write once, as above...
    # with engine.begin() as conn: conn.execute(..., flat)

@flow
def process_and_write_chunked(records: list[dict], batch_size: int = 5_000):
    chunks = [list(c) for c in batched_iterable(records, batch_size)]
    processed_per_chunk = process_chunk.map(chunks)
    with tags("db_write"):
        write_batch(processed_per_chunk)

Operational tips - Concurrency limits: If multiple flow runs may hit your DB, use tag-based concurrency limits on the “db_write” tag to cap concurrent batch writes. See Tag-based concurrency limits and Global concurrency limits. - Retries and backoff: Configure retries on the batch write task for transient DB errors. Docs: Retries.

- Mapping/reduce pattern: Prefect will automatically resolve mapped futures passed to a downstream task, enabling elegant “reduce” patterns. See Run work concurrently and this overview of mapping at scale: Task mapping at scale. A couple of clarifying questions to tailor this further: - Which database are you targeting (Postgres, Snowflake, BigQuery, etc.)? - Are you doing inserts only or upserts/merges? - How large can a “batch” get and do you have memory constraints? - Do you need a single transaction across the whole batch, or is per-chunk commit acceptable? With those details, I can share a dialed-in example (including SQL and driver-specific bulk operations).

Got it — you’re turning one “initial” record into a bundle of related objects that must be persisted across multiple tables. The key is to keep record-level processing parallelized, then perform a single, centralized, multi-table write in a single database transaction to preserve referential integrity. Recommended strategy in Prefect 3.x - Map for processing, reduce for writing: - Map a pure processing task over the input records; return a structured payload that contains rows for each target table. - Pass the list of mapped results to a single downstream write task. Prefect will resolve the futures and pass you a concrete list. - Inside the write task, open one DB transaction and write all tables in the correct order. This is the only way to guarantee ACID across multiple tables in Prefect (transactions cannot span multiple tasks). - Chunk if needed: - For large inputs, chunk first (e.g., 5k–50k records per chunk), map a process_chunk task, then reduce to one write task that flattens and writes everything. Or, if volumes are huge, process/write per chunk with one transaction per chunk. - Use deterministic keys: - If possible, generate stable IDs (e.g., UUIDs) during processing for parent/child rows. This avoids “insert-then-fetch-ID” loops and simplifies child inserts. - If you must use DB-generated IDs, do an INSERT ... RETURNING for parents, build a mapping, then insert children. - Retries and concurrency: - Put retries with exponential backoff on the write task. - Use tag-based concurrency limits on your write task (e.g., “db_write”) to protect the database from overload if multiple flows run concurrently. - Docs: Retries, Tag-based concurrency limits, Run work concurrently. Example: multi-table write with a single transaction (Postgres + SQLAlchemy Core) ``` from typing import Dict, List from prefect import flow, task, tags from prefect.tasks import exponential_backoff # Shape returned by processing: rows for each table # { # "parents": [{"id": "...", "name": ...}, ...], # "children_a": [{"id": "...", "parent_id": "...", ...}, ...], # "children_b": [{"id": "...", "parent_id": "...", ...}, ...], # } @task def process_record(rec: dict) -> Dict[str, List[dict]]: # Derive deterministic IDs (e.g., UUID5 from natural keys) to avoid round-trips parent_id = rec["id"] # or your own uuid generation parent_row = { "id": parent_id, "name": rec["name"].strip(), } children_a = [ {"id": f"{parent_id}-A-{i}", "parent_id": parent_id, "attr": v} for i, v in enumerate(rec.get("attrs_a", [])) ] children_b = [ {"id": f"{parent_id}-B-{i}", "parent_id": parent_id, "attr": v} for i, v in enumerate(rec.get("attrs_b", [])) ] return {"parents": [parent_row], "children_a": children_a, "children_b": children_b} @task def coalesce(bundles: List[Dict[str, List[dict]]]) -> Dict[str, List[dict]]: # Flatten and deduplicate parents by id parents, children_a, children_b = [], [], [] for b in bundles: parents.extend(b["parents"]) children_a.extend(b["children_a"]) children_b.extend(b["children_b"]) # Deduplicate parent rows by id (last-write-wins or merge fields as you need) seen = {} for p in parents: seen[p["id"]] = p parents = list(seen.values()) return {"parents": parents, "children_a": children_a, "children_b": children_b} @task( retries=3, retry_delay_seconds=exponential_backoff(2), ) def write_all(tables: Dict[str, List[dict]]): from sqlalchemy import create_engine, text engine = create_engine("postgresql+psycopg2://user:pass@host:5432/db") parents = tables["parents"] children_a = tables["children_a"]

children_b = tables["children_b"] # Do everything atomically with engine.begin() as conn: # Upsert parents first if parents: conn.execute( text(""" INSERT INTO parents (id, name) VALUES (:id, :name) ON CONFLICT (id) DO UPDATE SET name = EXCLUDED.name """), parents, ) # Then children A if children_a: conn.execute( text(""" INSERT INTO children_a (id, parent_id, attr) VALUES (:id, :parent_id, :attr) ON CONFLICT (id) DO UPDATE SET attr = EXCLUDED.attr """), children_a, ) # Then children B if children_b: conn.execute( text(""" INSERT INTO children_b (id, parent_id, attr) VALUES (:id, :parent_id, :attr) ON CONFLICT (id) DO UPDATE SET attr = EXCLUDED.attr """), children_b, ) @flow def pipeline(records: List[dict], chunk_size: int | None = None): if chunk_size: # Optional: chunk first if records are large from prefect.utilities.collections import batched_iterable chunks = [list(c) for c in batched_iterable(records, chunk_size)] processed_per_chunk = process_record.map([r for chunk in chunks for r in chunk]) # map flat tables = coalesce(processed_per_chunk) else: processed = process_record.map(records) tables = coalesce(processed) # Limit concurrency on DB writes if multiple runs may overlap with tags("db_write"): write_all(tables) ``` Notes and options - If you must use DB-generated surrogate IDs: - Insert parents with INSERT ... RETURNING (or SQLAlchemy session.add_all + flush) to collect a mapping from natural key → id, then populate child rows with those IDs before inserting. - Keep this inside the same engine.begin() context. - Data warehouses (Snowflake/BigQuery/Redshift): - Prefer staging + MERGE. E.g., write all processed outputs to staging tables/files (CSV/Parquet), then run MERGE statements for each target table in a single task (and transaction where supported). This is often much faster and more scalable than row-by-row upserts. - Don’t split one multi-table transaction across multiple tasks: - Prefect tasks are separate units of work; you cannot span a single DB transaction across tasks. For all-or-nothing multi-table semantics, keep it in one write task. - Concurrency/rate limiting and resiliency: - Add tag-based concurrency limits to “db_write” if multiple deployments run concurrently. - Use targeted retries (e.g., only retry on transient errors) via a custom retry_condition_fn if you need finer control. - Docs: Retries, Concurrency limits. If you can share: - Which database (Postgres, Snowflake, BigQuery, etc.)? - Whether you can generate deterministic IDs vs. relying on DB-generated IDs? - Expected batch size and latency requirements? I can tailor the write task and SQL (including MERGE patterns) to your stack.

That warning comes from urllib3/requests when you’re making more concurrent HTTP requests to the same host than the connection pool is sized for (default is 10). Extra connections get created and then discarded, which is noisy and less efficient. There are three levers to fix it: 1) Reuse a configured Session with a larger pool - Create a single reusable

requests.Session

per process/thread with a bigger pool and pool_block=True (so callers wait instead of oversubscribing). - Make sure you fully consume/close responses so connections return to the pool.

import threading
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

_thread_local = threading.local()

def get_session():
    if not hasattr(_thread_local, "session"):
        s = requests.Session()
        retry = Retry(
            total=3,
            backoff_factor=0.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET", "POST", "PUT", "DELETE", "PATCH"],
        )
        adapter = HTTPAdapter(
            pool_connections=100,   # total pools
            pool_maxsize=100,       # per-host concurrent keep-alive connections
            max_retries=retry,
            pool_block=True,        # block instead of creating/discarding extras
        )
        s.mount("http://", adapter)
        s.mount("https://", adapter)
        _thread_local.session = s
    return _thread_local.session

def http_get(url, **kwargs):
    session = get_session()
    # Ensure the response is closed or fully read so the connection is returned to the pool
    with session.get(url, timeout=30, **kwargs) as r:
        r.raise_for_status()
        return r.json()

Notes: - Don’t create a new Session per request; reuse the configured one. - If you use

stream=True

, read to EOF and close, or the connection won’t be returned to the pool. 2) Reduce concurrency from Prefect - If lots of tasks call the same host concurrently, cap them with a tag-based concurrency limit so you don’t exceed your pool. - CLI (verified):

prefect concurrency-limit create http_outbound 20

- Then tag the tasks that make HTTP calls:

from prefect import task, tags

@task
def fetch(url):
    return http_get(url)

# when calling the task
with tags("http_outbound"):
    fetch.map(urls)

- Docs: Tag-based concurrency limits and Global concurrency limits. You can also limit local parallelism:

from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def my_flow(...):
    ...

3) Match pool sizing to reality - If your upstream or networking stack limits connections (e.g., 50 per client), set

pool_maxsize

and Prefect concurrency to stay within that. - If you’re running in multiple pods/workers, remember the limit applies per process; multiply accordingly. If you’re using an async client instead of requests: - httpx: configure limits with

httpx.Limits(max_keepalive_connections=..., max_connections=...)

and reuse a single Client. - aiohttp: reuse a

ClientSession

with a

TCPConnector(limit=..., limit_per_host=...)

. If you share a bit about how you’re calling that 172.19.8.47 endpoint (requests vs httpx/aiohttp, per-task or shared session), I can tailor the exact code pattern.

You can do this reliably with a “map → reduce → single transaction” pattern: - Map: process each input record into a bundle with a parent payload and child payloads that reference the parent by a stable client key (e.g., external_id), not the DB id. - Reduce: collect all bundles into a single structure. - Single write task: in one DB transaction, insert/upsert parents with INSERT … RETURNING (or ORM flush) to get DB-generated IDs, build a key→id map, then insert children using the mapped parent ids. Key point: you cannot span a database transaction across multiple Prefect tasks. Do the multi-table write in a single task to keep it atomic. Example (Postgres + SQLAlchemy Core; mapped results feed one write task) ``` from typing import Dict, List from prefect import flow, task, unmapped, tags from prefect.tasks import exponential_backoff # Mapped task: return parent/children, children reference parent by a client key @task def process_record(rec: dict) -> Dict[str, List[dict]]: parent = { "external_id": rec["external_id"], # unique natural key for upsert/id map "name": rec["name"], } children_a = [ {"parent_external_id": rec["external_id"], "attr": v} for v in rec.get("attrs_a", []) ] children_b = [ {"parent_external_id": rec["external_id"], "attr": v} for v in rec.get("attrs_b", []) ] return {"parents": [parent], "children_a": children_a, "children_b": children_b} @task def coalesce(bundles: List[Dict[str, List[dict]]]) -> Dict[str, List[dict]]: # Flatten and deduplicate parents by external_id parents, children_a, children_b = [], [], [] for b in bundles: parents.extend(b["parents"]) children_a.extend(b["children_a"]) children_b.extend(b["children_b"]) dedup = {} for p in parents: dedup[p["external_id"]] = p return {"parents": list(dedup.values()), "children_a": children_a, "children_b": children_b} @task( retries=3, retry_delay_seconds=exponential_backoff(2), ) def write_all(tables: Dict[str, List[dict]]): # Define your SQLAlchemy tables elsewhere and import them here. # Below shows raw SQL for clarity. from sqlalchemy import create_engine, text engine = create_engine("postgresql+psycopg2://user:pass@host:5432/db") parents = tables["parents"] children_a = tables["children_a"] children_b = tables["children_b"] if not parents and not children_a and not children_b: return with engine.begin() as conn: # 1) Upsert parents and RETURNING (id, external_id) to build id map id_map = {} if parents: res = conn.execute( text(""" INSERT INTO parents (external_id, name) VALUES (:external_id, :name) ON CONFLICT (external_id) DO UPDATE SET name = EXCLUDED.name RETURNING id, external_id """), parents, ) rows = res.fetchall() id_map = {row.external_id: row.id for row in rows} # 2) Prepare children with db-generated parent_id def attach_parent_id(child_rows): out = [] for r in child_rows: pid = id_map.get(r["parent_external_id"]) if pid is None: # Defensive: if upsert didn’t return, fetch mapping (rare for PG), # or raise. Usually not needed for Postgres. raise ValueError(f"Missing parent id for {r['parent_external_id']}") nr = dict(r) nr["parent_id"] = pid nr.pop("parent_external_id", None) out.append(nr) return out ins_children_a = attach_parent_id(children_a) if children_a else [] ins_children_b = attach_parent_id(children_b) if children_b else [] # 3) Insert/upsert children if ins_children_a:

conn.execute( text(""" INSERT INTO children_a (parent_id, attr) VALUES (:parent_id, :attr) ON CONFLICT DO NOTHING """), ins_children_a, ) if ins_children_b: conn.execute( text(""" INSERT INTO children_b (parent_id, attr) VALUES (:parent_id, :attr) ON CONFLICT DO NOTHING """), ins_children_b, ) @flow def pipeline(records: List[dict]): processed = process_record.map(records) tables = coalesce(processed) # Optional: tag to cap concurrent writes across runs with tags("db_write"): write_all(tables) ``` Notes and alternatives - ORM approach: add all Parent objects, session.flush() to populate DB-generated ids, build the map in-memory, then create/save Child objects. Keep everything within a single session/transaction. - MySQL or warehouses without robust RETURNING: - Upsert parents (ON DUPLICATE KEY UPDATE), then SELECT id, external_id WHERE external_id IN (...) to build the map, then insert children. Keep it in one transaction. - Idempotency: ensure parents have a unique key (e.g., external_id) and use upserts so retries don’t duplicate data. - Large batches: chunk parent upserts (e.g., 5k–50k rows) then fetch mapping for that chunk and insert its children before moving to the next chunk, all within the same task. Prefect specifics - Passing mapped futures directly into `coalesce`/`write_all` is supported; Prefect resolves them into lists for you. - Use retries on the write task and tag-based concurrency limits if multiple deployments hit the DB at once. - Docs: Retries, Concurrency limits, Run work concurrently. If you share your database (Postgres/MySQL/Snowflake/etc.) and whether you’re using Core or ORM, I can tailor the insert/upsert and RETURNING/SELECT mapping code precisely.

Great question. In Prefect 3.x, a process worker can run your flow from a pre-built artifact by using “pull” steps in your deployment to download and use that artifact at runtime. Here’s a clean, repeatable setup. Assumptions - You’re using Prefect 3.x. - Your CI builds an artifact (e.g., a zip/tar of your flow code) and uploads it to stable remote storage (e.g., s3://my-bucket/artifacts/myapp/latest.zip). - The process worker’s host already has Python and your dependencies installed. With a process worker, the code runs in the host environment; if you want dependencies baked into the runtime, consider the Docker worker instead. Option A: Prefect YAML with pull steps (recommended for pre-built artifacts) 1) Create a process work pool and start a worker

prefect work-pool create my-process-pool --type process
prefect worker start --pool my-process-pool

2) Initialize a project and define your deployment’s pull steps

prefect init --name my-project

Edit prefect.yaml so your deployment downloads and unpacks the artifact, then points to the flow entrypoint. Example for an S3-hosted zip:

name: my-project
prefect-version: ">=3.0.0"

deployments:
- name: etl-prod
  description: Runs ETL from a pre-built artifact
  entrypoint: app/flows/etl.py:etl   # module path inside your artifact
  work_pool:
    name: my-process-pool
  pull:
  - prefect.deployments.steps.pull.pull_from_remote_storage:
      # Use a stable alias like "latest.zip" that CI updates each build
      url: "<s3://my-bucket/artifacts/myapp/latest.zip>"
      target: "."
  - prefect.deployments.steps.utility.run_shell_script:
      # Adjust unzip/tar command to match your artifact
      script: |
        set -e
        unzip -o latest.zip -d .
        rm -f latest.zip
  - prefect.deployments.steps.pull.set_working_directory:
      directory: .   # or the folder created by extraction (e.g., ./app)

3) Deploy

prefect deploy ./app/flows/etl.py:etl --name etl-prod

At runtime, the process worker will: - Download your artifact - Extract it - Set the working directory - Import and run the flow via your entrypoint Notes - Ensure the worker environment has your flow’s dependencies installed ahead of time. With a process worker, Prefect does not create an isolated environment by default. - If using S3/GCS/HTTP, the worker must have credentials available (e.g., AWS env vars) so it can read the artifact URL. - If your artifact is a directory tree instead of a zip, you can skip the unzip step and just use pull_from_remote_storage + set_working_directory. Option B: Deploy via Python using remote source (no custom YAML) If you store your flow code in remote storage as plain files (not zipped), Prefect can automatically configure the pull:

from prefect import flow

@flow
def etl():
    ...

# The 'source' can be an s3://, gs://, or other fsspec URL that points to your code root
etl.from_source(
    source="<s3://my-bucket/myapp/code/>",           # a folder containing your code
    entrypoint="app/flows/etl.py:etl"
).deploy(
    name="etl-prod",
    work_pool_name="my-process-pool",
)

This creates a deployment that will pull from the remote source at runtime without you managing custom pull steps. Key docs - Prefect YAML and pull steps: prefect.yaml - Store flow code remotely: Store flow code for deployments - Deploy via Python: Deploy via Python API - Pull steps API: prefect.deployments.steps.pull - Process worker reference: prefect.workers.process Quick checklist - Artifact built and uploaded by CI to a stable URL - Worker host has Python + dependencies installed

- Deployment pull steps download and extract the artifact - Entrypoint points to your flow inside the artifact - Process worker has credentials to read the artifact (e.g., AWS creds) If you can share: - Where your artifact lives (S3, GCS, etc.) - Artifact format (zip, tar, plain directory) - How you handle dependencies (pre-installed vs. need to install on the worker) …I can tailor the exact prefect.yaml (or Python) snippet for your setup.

Great question. In Prefect 3.x you can configure a deployment so a Process worker pulls a pre-built artifact (zip, wheel, etc.) and runs it without cloning your repo at runtime by using pull steps in your deployment. There are two common patterns: 1) Zip/tar archive of your project (run from the extracted code) - Store a zip/tar of your flow project in remote storage (S3/GCS/HTTP/etc.) - Pull the archive to the worker at runtime, extract it, set the working directory, then run the flow from the local files Example prefect.yaml:

# prefect-version is optional; shown for clarity
prefect-version: "3.*"
name: prebuilt-artifact-project

deployments:
- name: etl-prod
  entrypoint: "flows/etl.py:etl"   # Path inside the extracted archive
  work_pool:
    name: process-pool
  pull:
    # 1) Download the pre-built artifact to the work directory
    - prefect.deployments.steps.pull.pull_from_remote_storage:
        url: "<s3://my-bucket/artifacts/etl-build-{{> env.BUILD_ID }}.zip"

    # 2) Extract it (adjust commands for tar.gz if needed)
    - prefect.deployments.steps.utility.run_shell_script:
        script: |
          rm -rf app && mkdir -p app
          unzip -q etl-build-*.zip -d app

    # 3) Set working directory to the extracted project
    - prefect.deployments.steps.pull.set_working_directory:
        directory: "app"

    # 4) Optionally install requirements from the artifact
    - prefect.deployments.steps.utility.pip_install_requirements:
        requirements_file: "requirements.txt"

2) Wheel (install and import the flow from the package) - Build a wheel of your code and store it in remote storage (or publish to a private index) - Pull and pip install the wheel at runtime, then import your flow from the installed package Example prefect.yaml:

prefect-version: "3.*"
name: wheel-artifact-project

deployments:
- name: etl-prod
  # Import the flow from the installed package instead of a repo file path
  entrypoint: "mypkg.flows:etl"
  work_pool:
    name: process-pool
  pull:
    # 1) Download your wheel
    - prefect.deployments.steps.pull.pull_from_remote_storage:
        url: "<s3://my-bucket/artifacts/mypkg-{{> env.BUILD_VERSION }}-py3-none-any.whl"

    # 2) Install it into the worker's environment
    - prefect.deployments.steps.utility.run_shell_script:
        script: "python -m pip install --no-deps ./mypkg-*.whl"

Notes and tips - Don’t add any git pull steps; only include the artifact pull/install steps so the worker never clones source. - Process workers run in the host Python environment, so make sure the environment has a compatible Python version and any base dependencies. If you need isolation, consider Docker/Kubernetes workers instead. - If your artifact is a tar.gz, swap the unzip step for tar, e.g.

tar -xzf artifact.tar.gz -C app

. - Python can import directly from zip files if added to PYTHONPATH; if you prefer that, replace extraction with a step that sets

PYTHONPATH

in

job_variables.env

or via a shell script export. How to create the pool and start the worker

# Create a process work pool (run once)
prefect work-pool create "process-pool" --type process

# Start a worker that polls your pool
prefect worker start --pool "process-pool"

How to deploy - Ensure the prefect.yaml is in your project root, then:

prefect deploy

Or do it programmatically if you prefer:

from prefect import flow, Flow

# If you want to wire source from code instead of YAML
Flow.from_source(
    "<s3://my-bucket/artifacts/etl-build-123.zip>",
    entrypoint="flows/etl.py:etl",
).deploy(
    name="etl-prod",
    work_pool_name="process-pool",
)

If you share: - where your artifact is stored (S3/GCS/HTTP/other), - its format (zip/tar/wheel), and - how you’d like dependencies handled, I can tailor the exact pull steps and entrypoint for your setup.

Got it — here are the likely bottlenecks in your current pattern and a faster, drop-in way to execute the same steps. Where the bottlenecks are - Calling .result() in the loop: -

freight_requests = f.result()

blocks the client until that whole search finishes and transfers the full list into memory. -

processed.result()

forces all processing to complete before you schedule the save — this removes most concurrency. - Saving one big blob synchronously: -

repository_instance.save_requests_pg.submit(processed.result())

waits for processing to finish, then performs a single large write. This can create head‑of‑line blocking and spikes in memory/DB load. - Too many small writes or too large single writes: - If you switch to per-item saves, you get many small round trips. If you save the whole thing at once, you get a single heavy transaction. Batching is faster and more stable. - Mixed I/O and CPU with a single execution model: - Search and DB are I/O-bound; processing may be CPU-bound. Using the same execution strategy for both can underperform. - Per-item side effects: - Publishing messages on the hot path can block. Make it asynchronous and/or batch where possible. - Hidden overheads: - Recreating clients or not reusing HTTP/DB connections inside tasks; excessive logging; unbounded concurrency against remote APIs. A faster pattern (no blocking, dynamic mapping, batched writes) Key changes: - Don’t call .result() in the loop — keep everything as futures. - Use dynamic mapping: map over a future that resolves to a list; Prefect will fan out at runtime. - Batch for DB writes. - Make the message publish a task and submit it so it doesn’t block. Example ``` from prefect import flow, task from prefect.futures import as_completed from prefect.task_runners import ThreadPoolTaskRunner # Utility: chunk a list into batches @task def chunk(items: list, size: int = 500) -> list[list]: return [items[i:i+size] for i in range(0, len(items), size)] # Make publish async so it doesn't block the loop @task def publish_refresh(rabbitmq_secret_block, tab_id, job_id, status="REFRESH"): publish_refresh_message(rabbitmq_secret_block, tab_id, job_id, status) @flow(name="timocom-etl", task_runner=ThreadPoolTaskRunner(max_workers=32)) def run_etl( logger, pg_utils, user, password, sanitized_search_parameters, search_parameters, ors_server_variable, price_calculation_server_variable, predictive_scoring_server_variable, here_maps_api_key_variable, rabbitmq_secret_block, save_batch_size: int = 500, ): # Initialize clients once timocom_instance = Timocom(logger, pg_utils, user, password) config = get_config( ors_server_variable, price_calculation_server_variable, predictive_scoring_server_variable, here_maps_api_key_variable, ) processor_instance = FreightRequestProcessor(config, logger) repository_instance = FreightRequestRepository(logger, pg_utils) # Kick off searches concurrently search_tasks = timocom_instance.create_search_tasks(sanitized_search_parameters) search_futures = timocom_instance.search.map(search_tasks) # list[Future[list]] # Stream results as searches finish — no .result() here for search_future in as_completed(search_futures): # Dynamically fan out processing over the list returned by search_future processed = processor_instance.process.map(search_future) # list[Future[item]] # Option A: Save per item (simple, but more round trips) # repository_instance.save_requests_pg.map(processed) # Option B (faster/stabler): batch saves processed_list = processed.gather() # Future[list[item]] batches = chunk(processed_list, save_batch_size) # Future[list[list[item]]] repository_instance.save_requests_pg.map(batches) # batch DB writes in parallel # Publish without blocking

publish_refresh.submit( rabbitmq_secret_block, search_parameters["tab_id"], search_parameters["job_id"], "REFRESH", )

Even faster for CPU-heavy processing
If `processor_instance.process` is CPU-bound (e.g., heavy scoring), run that part with a process pool for true parallelism, while keeping search/DB on threads.

- Wrap CPU processing in a subflow with ProcessPoolTaskRunner.
- Submit that subflow per search batch.

Sketch:

from prefect import flow from prefect.task_runners import ProcessPoolTaskRunner @flow(task_runner=ProcessPoolTaskRunner(max_workers=8)) def process_batch(processor_instance, items: list): # Map across items in processes return processor_instance.process.map(items).gather() @flow(task_runner=ThreadPoolTaskRunner(max_workers=32)) def main(...): ... for search_future in as_completed(search_futures): processed_list = process_batch.submit(processor_instance, search_future) # Future[list] batches = chunk(processed_list, 500) repository_instance.save_requests_pg.map(batches) publish_refresh.submit(...) ``` Tuning tips - Batch size: 200–1000 is a good starting range for DB upserts; use COPY/execute_batch/execute_values where possible inside

save_requests_pg

. - Concurrency: - ThreadPoolTaskRunner max_workers: tune based on I/O latency and API limits (8–64). - ProcessPoolTaskRunner max_workers: number of CPU cores for CPU-heavy steps. - External services: reuse HTTP sessions/DB connections inside tasks; set timeouts; add backoff; apply concurrency limits to avoid rate limiting. - Logging: reduce per-item logging; log per batch. - Memory: avoid holding entire datasets on the client; keep lists as futures and operate on them in tasks/subflows. If you share: - Typical sizes (search count, items per search), - Whether

process

is CPU or I/O-bound, - How

save_requests_pg

writes (single insert vs bulk), I can tailor batch sizes, worker counts, and a concrete

save_requests_pg

bulk-insert pattern to squeeze more throughput.

You’re right that the way futures are being passed around is the likely culprit. The fix is to ensure the DB-write task receives concrete values at the right granularity, without blocking the whole pipeline. Key points - Passing a Prefect future into a Prefect task is OK — Prefect will wait for it before the task runs. Passing a future into plain Python (non-task) code is not. - Avoid calling

.result()

inside the flow unless you intend to block; use

.gather()

to collapse lists of futures into a single future-of-list, and let Prefect schedule downstream tasks. - Choose the right save pattern: - If your DB task saves one item: map over processed items. - If it saves a batch: gather, then chunk, then map over batches. Fix 1: If your save task expects a batch (list of items)

from prefect import task
from prefect.futures import as_completed

@task
def chunk(items: list, size: int = 500) -> list[list]:
    return [items[i:i+size] for i in range(0, len(items), size)]

for f in as_completed(search_futures):
    # Dynamically map processing over the list returned by f
    processed = processor_instance.process.map(f)      # list[Future[item]]

    # Collapse to a single future-of-list without blocking the flow
    processed_list = processed.gather()                # Future[list[item]]

    # Optional: batch to stabilize DB load
    batches = chunk.submit(processed_list, size=500)   # Future[list[list[item]]]

    # Map DB writes over batches; Prefect waits for each batch future before running
    repository_instance.save_requests_pg.map(batches)

    # Publish without blocking
    publish_refresh_message(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
    )

Fix 2: If your save task expects a single item

for f in as_completed(search_futures):
    processed = processor_instance.process.map(f)  # list[Future[item]]
    repository_instance.save_requests_pg.map(processed)

Fix 3: If you must “save all processed items from this search” in one call You can submit the save task with a future-of-list directly; Prefect will wait for it:

for f in as_completed(search_futures):
    processed = processor_instance.process.map(f)   # list[Future[item]]
    processed_list = processed.gather()             # Future[list[item]]
    repository_instance.save_requests_pg.submit(processed_list)

What to double-check -

save_requests_pg

is a Prefect task (

@task

) and you’re calling it with

.submit(...)

or

.map(...)

(not as a plain function). - Any helper like

chunk

that takes futures must also be a Prefect task, or you must call

.result()

before invoking a plain Python function. - Look for upstream failures in logs; Prefect will cancel downstream work if an upstream task failed. The “Execution was cancelled by the runtime environment” message often indicates upstream failure or the worker shutting down before the task started. Optional hardening - Add retries/timeouts to the DB task to mitigate transient DB issues:

@task(retries=3, retry_delay_seconds=5, timeout_seconds=60)
def save_requests_pg(batch: list[dict]):
    ...

- If processing is CPU-heavy, run processing with a ProcessPoolTaskRunner and keep DB I/O on threads. If you can share the

save_requests_pg

task signature (expects one item vs batch) and a small snippet of the current loop, I’ll tailor the exact wiring to eliminate the cancellation.

Got it. If

save_requests_pg

expects a batch and your

processed = processor_instance.process.map(search_future)

already yields batches, the simplest fix is to stop resolving futures and just map the save across those batch futures. Prefect will wait for each processed batch before running the corresponding DB write. Do this:

for search_future in as_completed(search_futures):
    # Each element of `search_future` is already a chunk
    processed_batches = processor_instance.process.map(search_future)  # list[Future[Batch]]

    # Write each processed batch as it becomes ready (no .result(), no gather)
    repository_instance.save_requests_pg.map(processed_batches)

    # Optionally publish without blocking the flow
    # (wrap as a @task and submit if you want it in the Prefect graph)
    publish_refresh_message(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
    )

Important checks - Ensure

save_requests_pg

is a Prefect task (decorated with

@task

) and you’re calling it with

.map(...)

or

.submit(...)

. - Don’t pass Prefect futures into plain Python functions. If you must use a helper that isn’t a task, resolve its input first (or convert that helper into a task). - That “Execution was cancelled by the runtime environment.” message can also occur if an upstream task failed; scan logs for any failures in the

process

tasks or earlier searches. If instead your

process

is item-level (not batch-level), you’ll want to gather to form one batch per search before saving:

for search_future in as_completed(search_futures):
    processed_items = processor_instance.process.map(search_future)  # list[Future[item]]
    batch_future = processed_items.gather()                          # Future[list[item]]
    repository_instance.save_requests_pg.submit(batch_future)        # single batched write

Tell me which of those two shapes you have (per-chunk vs per-item process), and I’ll tailor the final snippet 1:1 with your signatures.

Perfect — since you already have a single processed batch per search, you just need to enforce the dependency “publish after save”. Preferred: make publish a Prefect task and add the save future as an upstream dependency (either via a dummy arg or wait_for)

from prefect import task
from prefect.futures import as_completed

@task
def publish_refresh(rabbitmq_secret_block, tab_id, job_id, status="REFRESH", _after=None):
    # `_after` is unused; it just enforces ordering
    publish_refresh_message(rabbitmq_secret_block, tab_id, job_id, status)

for search_future in as_completed(search_futures):
    # process returns a single batch for this search
    processed_batch = processor_instance.process.submit(search_future)   # Future[Batch]

    # save expects a batch
    save_future = repository_instance.save_requests_pg.submit(processed_batch)

    # Option A (works in all cases): pass save_future as a dummy argument to enforce ordering
    publish_refresh.submit(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
        _after=save_future,       # ensures publish runs only after save completes successfully
    )

    # Option B (if you prefer and your version supports it):
    # publish_refresh.submit(
    #     rabbitmq_secret_block,
    #     search_parameters["tab_id"],
    #     search_parameters["job_id"],
    #     "REFRESH",
    #     wait_for=[save_future],
    # )

If you can’t make

publish_refresh

a Prefect task and must call the plain function, block on the save future first:

save_future = repository_instance.save_requests_pg.submit(processed_batch)
save_future.wait()  # or .result()
publish_refresh_message(rabbitmq_secret_block, tab_id, job_id, "REFRESH")

Notes - Passing a Prefect future as an argument to another Prefect task automatically creates a dependency; Prefect will run the downstream task only after the upstream completes successfully. - Using the dummy

_after

arg keeps publish from running concurrently with the DB write without changing your publish logic.

You’re almost there. Since

process.map(search_future)

returns a list of futures (one per item) but

save_requests_pg

expects a single batch (list of items), you should gather those item futures into a single future-of-list before submitting the save. Then make publish depend on the save future. Key points - Map over items: process.map(search_future) -> list[Future[item]] - Gather into one batch: processed_items.gather() -> Future[list[item]] - Submit save with the batch future: Prefect will resolve the batch before running - Enforce “publish after save” by making publish a Prefect task that takes the save future as a dummy arg (or a wait parameter) Drop-in pattern

for search_future in as_completed(search_futures):
    <http://logger.info|logger.info>(f"Start processing: {datetime.now()}")

    # 1) Process each item (mapping yields list of futures)
    processed_items = processor_instance.process.map(search_future)   # list[Future[item]]

    # 2) Collapse to a single future-of-list (non-blocking)
    processed_batch = processed_items.gather()                        # Future[list[item]]

    # 3) Save expects a batch (list of items)
    save_future = repository_instance.save_requests_pg.submit(processed_batch)

    <http://logger.info|logger.info>("Requests data persistence scheduled.")  # scheduled, not completed yet

    # 4) Publish only after save finishes
    # Ensure publish_refresh_message is a @task that accepts an extra arg to enforce ordering:
    # def publish_refresh_message(..., _after=None): ...
    publish_refresh_message.submit(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
        _after=save_future,   # forces dependency without blocking the flow runner
    )

Notes - Don’t pass a raw list[Future] to `save_requests_pg.submit`; either gather (as above) or change

save_requests_pg

to accept per-item and use map. Since you want a single batch, gather is the right choice. - The

_after=save_future

approach requires

publish_refresh_message

to accept an

_after

parameter (unused). Alternatively, pass

save_future

as a dummy argument the task already accepts, or create a small wrapper task that accepts

wait_for: Any | None = None

. - If you truly need the “completed” log after the save finishes, make that log a tiny Prefect task that depends on

save_future

, rather than logging immediately in the flow code.

You’re seeing that error because

PrefectFutureList

doesn’t have a

.gather()

method in Prefect 3.x. You don’t need it here anyway. Since: -

search_future

-> list[Item] -

process.map(search_future)

-> list[Future[ProcessedItem]] (already item-level) -

save_requests_pg

expects a batch list[ProcessedItem] You can pass the list of futures directly to the save task. Prefect will resolve nested futures in task inputs before running the task. Drop-in fix (keep map, then submit, and publish only after save):

for search_future in as_completed(search_futures):
    <http://logger.info|logger.info>(f"Start processing: {datetime.now()}")

    # Map over items -> list of futures
    processed_items = processor_instance.process.map(search_future)  # PrefectFutureList[ProcessedItem]

    # Pass the future list directly; Prefect resolves it to list[ProcessedItem] at runtime
    save_future = repository_instance.save_requests_pg.submit(processed_items)

    <http://logger.info|logger.info>("Requests data persistence scheduled.")
    <http://logger.info|logger.info>(f"Publish message: {datetime.now()}")

    # Ensure publish runs after save completes
    # Option 1: dummy arg to enforce dependency (preferred if your task accepts it)
    publish_refresh_message.submit(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
        _after=save_future,  # unused param to enforce dependency
    )

    # Option 2: if your version supports it, you can also do:
    # publish_refresh_message.submit(..., wait_for=[save_future])

Alternative if you really want a single “future-of-list” (gather semantics) before saving:

from prefect import task

@task
def collect(items):
    # items arrives as list[ProcessedItem] since Prefect resolves futures
    return items

processed_items = processor_instance.process.map(search_future)  # list[Future[item]]
batch_future = collect.submit(processed_items)                  # Future[list[item]]
save_future = repository_instance.save_requests_pg.submit(batch_future)
publish_refresh_message.submit(..., _after=save_future)

Key points - Passing a list of futures into a Prefect task is valid; Prefect resolves them to concrete values before executing the task. - Remove `.gather()`; it’s not needed and causes the AttributeError you saw. - Make sure

publish_refresh_message

is a Prefect task. If it’s a plain function, block after save with

save_future.result()

before calling it.

Thanks for the detail — the crash is almost certainly coming from how the list of futures is being passed to the DB task. In Prefect 3.x, a Prefect task will resolve Prefect futures in its inputs, but passing a PrefectFutureList directly can be brittle in some contexts and lead to cancellations if upstream items are not ready or one failed. Given your constraints: - search_future -> list[item] - process.map(search_future) -> list[Future[processed_item]] (already the single batch you want) - save_requests_pg expects list[processed_item] (a concrete batch) - publish must run strictly after save Make it robust by inserting a tiny “collector” task that turns the list of futures into a single future-of-list, then chain save and publish. This avoids the PrefectFutureList edge case and gives a clean dependency chain. Drop-in fix

from prefect import task
from prefect.futures import as_completed

# 1) Collect list[Future[item]] -> Future[list[item]]
@task
def collect_batch(items):
    # Prefect resolves futures in inputs; you receive concrete items here
    return items

# 2) Make publish a task so we can depend on the save
@task
def publish_refresh(rabbitmq_secret_block, tab_id, job_id, status="REFRESH", _after=None):
    publish_refresh_message(rabbitmq_secret_block, tab_id, job_id, status)

for search_future in as_completed(search_futures):
    <http://logger.info|logger.info>(f"Start processing: {datetime.now()}")

    # Map per item -> list[Future[processed_item]]
    processed_items = processor_instance.process.map(search_future)

    # Collapse to a single future-of-list for the batch
    processed_batch = collect_batch.submit(processed_items)          # Future[list[processed_item]]

    # Save expects a batch
    save_future = repository_instance.save_requests_pg.submit(processed_batch)

    <http://logger.info|logger.info>("Requests data persistence scheduled.")
    <http://logger.info|logger.info>(f"Publish message: {datetime.now()}")

    # Publish strictly after save completes
    publish_refresh.submit(
        rabbitmq_secret_block,
        search_parameters["tab_id"],
        search_parameters["job_id"],
        "REFRESH",
        _after=save_future,   # unused param to enforce dependency
    )

Why this fixes the crash - We avoid passing a PrefectFutureList directly into the save task; instead,

collect_batch

produces a single Future[list[processed_item]] that is safe to pass to

save_requests_pg.submit

. - Dependencies are explicit: publish waits on save via

_after=save_future

. - No blocking in the flow code; Prefect schedules each step when its upstreams finish. If you don’t want a separate collector task - You can change the DB task signature to accept an Iterable and rely on Prefect to resolve inputs, but in practice the collector pattern is the most reliable when mixing map and submit. Sanity checks - Ensure

processor_instance.process

,

repository_instance.save_requests_pg

, and

publish_refresh

are Prefect tasks (decorated with

@task

) — not plain functions. - If publish must remain a plain function, call it only after blocking for the save to complete:

save_future = repository_instance.save_requests_pg.submit(processed_batch)
  save_future.result()  # waits for DB write
  publish_refresh_message(...)

- If you still see “Execution was cancelled by the runtime environment.”, check the upstream

process

task logs for failures; Prefect cancels downstream work when an upstream fails or if the worker process shuts down. If you can share the task decorators/signatures for

process

,

save_requests_pg

, and

publish_refresh_message

, I can double-check the exact wiring.

Short answer: You’re trying to submit new work to a thread/process pool after it’s already been shut down. In Python, this is raised by

concurrent.futures

when an executor has been closed and you call

submit()

again. In Prefect 3.x, this usually means the flow’s task runner (e.g.,

ThreadPoolTaskRunner

) has already shut down when something tries to schedule another task. The task runner is started when the flow begins and is shut down as the flow exits; after that point, no new tasks can be submitted. Common triggers in Prefect: - The flow returns, then code tries to submit more tasks afterward (e.g., in cleanup code or background threads). - Futures are still being used to trigger new work after the flow has ended. - Asynchronous or background callbacks fire late and try to schedule additional tasks after runner shutdown. - Re-using objects tied to a previous flow run’s task runner. How to fix: - Only call

task.submit(...)

while the flow is actively running (before the function returns). - Resolve or wait on futures before the flow ends:

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

  @task
  def work(i): 
      return i * i

  @flow
  def my_flow():
      futs = [work.submit(i) for i in range(5)]
      # Ensure all scheduled work completes before the flow exits
      wait(futs)  # or: results = [f.result() for f in futs]

- If you need to schedule more work “later,” wrap that logic in another flow/subflow that runs after the first flow completes rather than trying to submit tasks from outside a running flow. - Avoid spawning background threads or callbacks that submit tasks after the flow finishes; make sure they complete inside the flow’s lifetime. Useful references: - Task runners in Prefect 3.x - Running work concurrently - Futures API (wait, as_completed, result) If you can share a small snippet or the stack trace showing where the submit happens, I can point to the exact spot that’s firing after shutdown.

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

map()

API on tasks, and you generally shouldn’t orchestrate from inside a task. Do your “mapping” at the flow level by submitting tasks over an iterable. If you meant Python’s built-in

map

, you can call it inside a task, but it will just run inline in that single task (no Prefect parallelism or visibility). Recommended pattern (fan-out/fan-in) in a flow:

from prefect import flow, task, unmapped
from prefect.task_runners import ThreadPoolTaskRunner

@task
def process_record(record, constant):
    return f"{record}-{constant}"

@task
def combine(results):
    # You can pass futures; Prefect will resolve them for you
    return ",".join(results)

@flow(task_runner=ThreadPoolTaskRunner())
def main(records):
    # Fan-out: submit one task per record
    futures = [process_record.submit(r, unmapped("same-for-all")) for r in records]

    # Fan-in: downstream task can take the list of futures directly
    combined = combine.submit(futures).result()
    return combined

if __name__ == "__main__":
    print(main(["a", "b", "c"]))

Notes - Use

unmapped(...)

for arguments you don’t want to vary across the mapped tasks. - For streaming results as they finish, you can iterate with

prefect.futures.as_completed(...)

. If you’re coming from Prefect 1.x’s

task.map

, that API was removed;

submit

+ list comprehensions/loops is the 2.x/3.x replacement. If you need nested orchestration, consider promoting the inner logic to a subflow instead of calling

submit

inside a task. Helpful docs: - Task runners (controls concurrency): Task runners - Futures utilities (e.g.,

as_completed

): Futures API -

unmapped

annotation: unmapped If you have a snippet you’re trying to convert to Prefect 3.x, paste it here and I’ll help adapt it.

You don’t need manual synchronization—just add a downstream “reduce” step that collects the mapped results, flattens them, and deduplicates. Here’s a simple, canonical pattern in Prefect 3.x:

from prefect import flow, task

@task
def produce(batch_id: int) -> list[int]:
    # Example: each run returns a list with some overlap
    return [batch_id, batch_id + 1, 42]

@task
def dedupe_all(lists_of_items: list[list[int]]) -> list[int]:
    unique = set()
    for lst in lists_of_items:
        unique.update(lst)
    return sorted(unique)

@flow
def main():
    # Run N producers in parallel
    futures = produce.map([1, 2, 3, 4])

    # Collect results from all mapped runs -> list[list[int]]
    lists = futures.result()

    # Reduce: flatten + dedupe
    unique_items = dedupe_all(lists)
    return unique_items

result = main()

Notes: -

task.map(...)

returns a `PrefectFutureList`; call

.result()

to aggregate mapped outputs into a single

list[...]

you can pass downstream. - If your producer can return a

set

directly, merging is even simpler: - Task returns

set[...]

- Merge with

set.union(*sets)

in the reducer task. - If items are not hashable (e.g., dicts), dedupe by a key: - Build a key (e.g., tuple of selected fields) and keep first-seen items. Example for unhashable items:

@task
def dedupe_by_key(list_of_lists: list[list[dict]]) -> list[dict]:
    seen = set()
    out = []
    for lst in list_of_lists:
        for item in lst:
            key = (item["id"], item.get("type"))
            if key not in seen:
                seen.add(key)
                out.append(item)
    return out

This pattern gives you a clean fan-out (map) → fan-in (reduce) pipeline without any extra synchronization primitives.