Great question. Moving 300k+ records fast on small hardware is all about minimizing bytes, maximizing parallelism (within limits), and avoiding rework. Here’s a battle-tested approach you can adapt to most /products APIs. Best-practice strategy - Prefer bulk exports if available - If the API offers async exports to object storage (S3/GCS) or NDJSON/CSV bulk endpoints, use them — they are almost always faster/cheaper than page-by-page. - Do incremental syncs instead of full loads - Use filters like

updated_after

,

since_id

, or

modified_at >= last_success

to pull only changes. Store the checkpoint so re-runs are tiny. - Use keyset/cursor pagination with a stable sort - Avoid offset pagination for large result sets; it gets slower and can skip/duplicate rows. Use

cursor

or

since_id

with an ascending, unique key (e.g.,

id

). - Maximize network efficiency - Request only the fields you need (e.g.,

fields=id,sku,price,updated_at

). - Use compression: add

Accept-Encoding: gzip

and/or

br

. - Use HTTP/2 if the server supports it — it improves throughput and resource use. - Use the largest allowed page size. - Parallelize safely - Async I/O with a small number of long-lived connections is highly efficient on modest hardware (e.g., httpx.AsyncClient). - Use a bounded number of concurrent requests (start ~16–64, then tune). A quick rule: optimal concurrency ≈ target_rps × p95_latency_seconds. - Respect rate limits; back off on 429/5xx and honor Retry-After. - Stream results to storage; do not hold in memory - Write records out incrementally (DB batch insert, Parquet row groups, or NDJSON lines) to keep memory flat. - Resumability and observability - Checkpoint last cursor/updated_at frequently so you can resume after failures. - Log throughput and error counts to tune concurrency and page size. - Deduplicate by primary key on write if your API can occasionally return overlaps (e.g., eventual consistency). Reference implementation (async, minimal hardware) - Pattern: a cursor queue with N async workers, a single writer, streaming to disk. Replace the writer with your DB/warehouse sink. - This example uses cursor-based pagination; adapt params/keys for your API. ``` import asyncio import json import os from typing import Optional, Tuple import httpx API_URL = os.environ.get("PRODUCTS_URL", "https://api.example.com/products") PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "250")) # use API max MAX_CONCURRENCY = int(os.environ.get("MAX_CONCURRENCY", "32")) FIELDS = os.environ.get("FIELDS", "id,sku,price,updated_at") # only what you need OUTPUT_PATH = os.environ.get("OUTPUT_PATH", "products.ndjson") TIMEOUT = httpx.Timeout(30.0, connect=10.0) HEADERS = { "Accept": "application/json", "Accept-Encoding": "gzip, br", "User-Agent": "product-sync/1.0", } class Stop: pass async def fetch_page(client: httpx.AsyncClient, cursor: Optional[str]) -> Tuple[list, Optional[str]]: params = {"limit": PAGE_SIZE, "fields": FIELDS} if cursor: params["cursor"] = cursor backoff = 1.0 while True: try: resp = await client.get(API_URL, params=params, headers=HEADERS) # Handle transient and rate-limit errors if resp.status_code in (429, 500, 502, 503, 504): retry_after = resp.headers.get("Retry-After") delay = float(retry_after) if retry_after else backoff await asyncio.sleep(delay) backoff = min(backoff * 2, 30.0) continue resp.raise_for_status() data = resp.json() # Adjust these keys to match your API's shape items = data.get("items") or data.get("data") or [] next_cursor = data.get("next_cursor") or (data.get("paging") or {}).get("next_cursor") return items, next_cursor

except (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.RemoteProtocolError, httpx.HTTPStatusError, httpx.TransportError): await asyncio.sleep(backoff) backoff = min(backoff * 2, 30.0) async def run_sync(start_cursor: Optional[str] = None): limits = httpx.Limits( max_connections=MAX_CONCURRENCY, max_keepalive_connections=MAX_CONCURRENCY, ) cursor_q: asyncio.Queue = asyncio.Queue() results_q: asyncio.Queue = asyncio.Queue(maxsize=10_000) # Seed the queue with the starting cursor await cursor_q.put(start_cursor) async with httpx.AsyncClient(http2=True, timeout=TIMEOUT, limits=limits) as client: async def worker(): while True: cursor = await cursor_q.get() if isinstance(cursor, Stop): cursor_q.task_done() break items, next_cursor = await fetch_page(client, cursor) for it in items: await results_q.put(it) if next_cursor: await cursor_q.put(next_cursor) cursor_q.task_done() async def writer(): count = 0 with open(OUTPUT_PATH, "w", encoding="utf-8") as f: while True: item = await results_q.get() if isinstance(item, Stop): results_q.task_done() break f.write(json.dumps(item, ensure_ascii=False) + "\n") count += 1 results_q.task_done() print(f"Wrote {count} products to {OUTPUT_PATH}") # Start workers and writer workers = [asyncio.create_task(worker()) for _ in range(MAX_CONCURRENCY)] writer_task = asyncio.create_task(writer()) # Wait until all cursors are processed await cursor_q.join() # Stop workers and writer for _ in workers: await cursor_q.put(Stop()) await asyncio.gather(*workers) await results_q.put(Stop()) await writer_task if name == "__main__": asyncio.run(run_sync()) ``` Notes on tuning - Page size: use the maximum the API allows for the best throughput per request. - Concurrency: start with 16–32. Increase until you see either 429s or diminishing throughput due to server limits. Keep CPU/memory flat even at high throughput because this is I/O-bound. - HTTP/2: keep http2=True; it lets many requests share connections efficiently. - Rate limiting: if the API has strict RPS limits, add a token bucket (e.g., aiolimiter) or a simple sleep between requests, and keep the concurrency high to hide latency. - Storage: for large pulls, write to Parquet via pyarrow or to the database in batches for faster downstream consumption. Using Prefect 3.x (optional but recommended) - Wrap this as a Prefect flow for retries, scheduling, logging, and checkpointing. You can store and retrieve your last successful

updated_after

from a Prefect Variable and run incrementally on a schedule. - If you need centralized throttling across multiple runs/agents, use Prefect concurrency limits (e.g., guard each request with a concurrency slot) so you never exceed the provider’s limits from multiple deployments. Questions to tailor this to your API - Which pagination model does your /products endpoint use? (cursor/next, since_id, offset/page) - Max page size allowed? - Published rate limits? Is HTTP/2 supported? Compression supported? - Can you filter by

updated_at

to do incremental syncs? - Any bulk export or async job endpoints? Share those details and I’ll adjust the code and parameters for maximum speed with your specific endpoint.

Got it — here’s a production-ready Prefect 3.x flow that pulls 300k+ products fast on modest hardware. It uses async I/O with bounded concurrency, streams results to disk (flat memory), handles retries/rate limits, and checkpoints progress in a Prefect Variable so runs are resumable. Notes - Assumes a cursor-based /products endpoint that returns

items

and

next_cursor

. If your API uses

updated_after

or a different shape, I can adapt this quickly. - Avoids creating a Prefect task per page (keeps overhead low); all parallelism is pure asyncio inside a single flow. - Uses

prefect.variables.Variable

to store the last

updated_after

(optional) so subsequent runs are incremental. Flow code ``` from future import annotations import asyncio import json import os from datetime import datetime, timedelta, timezone from typing import Optional, Tuple, List, Dict, Any import httpx from prefect import flow, get_run_logger from prefect.variables import Variable # Tunables (can also be provided as flow parameters) DEFAULT_PAGE_SIZE = int(os.environ.get("PAGE_SIZE", "250")) # use API max DEFAULT_MAX_CONCURRENCY = int(os.environ.get("MAX_CONCURRENCY", "32")) DEFAULT_FIELDS = os.environ.get("FIELDS", "id,sku,price,updated_at") DEFAULT_OUTPUT_PATH = os.environ.get("OUTPUT_PATH", "products.ndjson") # Optional incremental sync checkpoint CHECKPOINT_VAR = os.environ.get("CHECKPOINT_VAR", "products_updated_after") CHECKPOINT_SKEW_SEC = int(os.environ.get("CHECKPOINT_SKEW_SEC", "60")) # re-fetch last minute TIMEOUT = httpx.Timeout(30.0, connect=10.0) HEADERS = { "Accept": "application/json", "Accept-Encoding": "gzip, br", "User-Agent": "prefect-products-sync/1.0", } def parse_ts(ts: str) -> datetime: # Robust-ish ISO8601 parse without extra deps # Handles 'Z' by normalizing to +00:00 if ts.endswith("Z"): ts = ts[:-1] + "+00:00" return datetime.fromisoformat(ts) def isoformat_utc(dt: datetime) -> str: if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") async def fetch_page( client: httpx.AsyncClient, api_url: str, page_size: int, fields: str, cursor: Optional[str], updated_after: Optional[str], ) -> Tuple[List[Dict[str, Any]], Optional[str]]: params = {"limit": page_size, "fields": fields} if cursor: params["cursor"] = cursor # Optional incremental filter if your API supports it if updated_after: params["updated_after"] = updated_after backoff = 1.0 while True: try: resp = await client.get(api_url, params=params, headers=HEADERS) if resp.status_code in (429, 500, 502, 503, 504): retry_after = resp.headers.get("Retry-After") delay = float(retry_after) if retry_after else backoff await asyncio.sleep(delay) backoff = min(backoff * 2, 30.0) continue resp.raise_for_status() data = resp.json() # Adjust these keys to match your API items = data.get("items") or data.get("data") or [] next_cursor = data.get("next_cursor") or (data.get("paging") or {}).get("next_cursor") return items, next_cursor except (httpx.ReadTimeout, httpx.ConnectTimeout, httpx.RemoteProtocolError, httpx.TransportError, httpx.HTTPStatusError): await asyncio.sleep(backoff) backoff = min(backoff * 2, 30.0) @flow(name="sync_products") async def sync_products_flow( api_url: str = "https://api.example.com/products", page_size: int = DEFAULT_PAGE_SIZE, max_concurrency: int = DEFAULT_MAX_CONCURRENCY, fields: str = DEFAULT_FIELDS, output_path: str = DEFAULT_OUTPUT_PATH, start_cursor: Optional[str] = None, use_incremental: bool = True, # set False for a full refresh ): """

High-throughput, low-memory product fetcher with resumability. - Cursor-based pagination with async concurrency - Optional incremental sync using updated_after checkpoint - Streams results to NDJSON (replace writer with your DB sink as needed) """ log = get_run_logger() # Determine incremental checkpoint updated_after: Optional[str] = None if use_incremental: last = Variable.get(CHECKPOINT_VAR, default=None) if isinstance(last, str) and last: # Apply skew so we re-fetch recent items to avoid gaps try: skewed = parse_ts(last) - timedelta(seconds=CHECKPOINT_SKEW_SEC) updated_after = isoformat_utc(skewed) except Exception: updated_after = last # fallback to raw value if parsing fails log.info(f"Using updated_after={updated_after!r} from Variable={CHECKPOINT_VAR!r}") limits = httpx.Limits(max_connections=max_concurrency, max_keepalive_connections=max_concurrency) max_seen_updated_at: Optional[datetime] = None cursor_q: asyncio.Queue = asyncio.Queue() results_q: asyncio.Queue = asyncio.Queue(maxsize=5_000) # Seed the cursor queue await cursor_q.put(start_cursor) async with httpx.AsyncClient(http2=True, timeout=TIMEOUT, limits=limits) as client: async def worker(): while True: cursor = await cursor_q.get() if cursor is StopAsyncIteration: cursor_q.task_done() break items, next_cursor = await fetch_page( client=client, api_url=api_url, page_size=page_size, fields=fields, cursor=cursor, updated_after=updated_after, ) for it in items: await results_q.put(it) if next_cursor: await cursor_q.put(next_cursor) cursor_q.task_done() async def writer(): nonlocal max_seen_updated_at count = 0 with open(output_path, "w", encoding="utf-8") as f: while True: item = await results_q.get() if item is StopAsyncIteration: results_q.task_done() break # Write as NDJSON line; replace this with your DB/warehouse sink as needed f.write(json.dumps(item, ensure_ascii=False) + "\n") count += 1 # Track max updated_at if present ua = item.get("updated_at") if isinstance(ua, str): try: ts = parse_ts(ua) if (max_seen_updated_at is None) or (ts > max_seen_updated_at): max_seen_updated_at = ts except Exception: pass # Light telemetry (avoid chatty logs) if count % (page_size * 10) == 0: log.info(f"Wrote {count:,} products so far...") log.info(f"Finished. Wrote {count:,} products to {output_path!r}") # Spin up workers workers = [asyncio.create_task(worker()) for _ in range(max_concurrency)] writer_task = asyncio.create_task(writer()) # Wait for all cursors to be processed await cursor_q.join() # Stop workers and writer for _ in workers: await cursor_q.put(StopAsyncIteration) await asyncio.gather(*workers) await results_q.put(StopAsyncIteration) await writer_task # Persist checkpoint for incremental runs if use_incremental and max_seen_updated_at: new_checkpoint = isoformat_utc(max_seen_updated_at)

Variable.set(CHECKPOINT_VAR, new_checkpoint, overwrite=True) log.info(f"Updated Variable {CHECKPOINT_VAR!r} to {new_checkpoint!r}") ``` How to run - As a one-off: call

sync_products_flow(api_url="<https://your.api/products>")

- On a schedule: create a deployment of this flow, then run on an interval (e.g., every 5–15 minutes). For maximal efficiency, keep

use_incremental=True

. Optional improvements - Replace the NDJSON writer with a database batch writer (e.g., Snowflake, Postgres, BigQuery) or Parquet via pyarrow. - If your org runs multiple deployments that hit the same API, add a global rate limit or concurrency guard so you never exceed provider limits. I can wire this up for your workspace. - If your API uses

updated_after

pagination instead of a cursor, we’ll tweak the fetch to pass a page token and iterate until no more results. Can you share: - The pagination model (cursor, since_id, updated_after + page_token)? - Max page size allowed and rate limits (RPS)? - The exact response shape for

items

and

next_cursor

? - Where you want to write the data (file, warehouse, object storage)? I’ll tailor the flow to your API and sink with those details.