Great question — and your FastAPI scenario is a perfect fit for Prefect 3.x background tasks. High-level difference - task.delay(): schedules a standalone task run to be executed by a Task Worker. Returns immediately with a PrefectDistributedFuture you can use to get the task_run_id, state, and result. Best for API-triggered, single-operation background jobs. - Wrapping a task in a flow: executes the task inside a flow run. This is ideal for multi-step orchestration, retries across tasks, mapping, etc. Typically used via a deployment and triggered with run_deployment. What task.delay() actually does - Creates a scheduled task run in the Prefect API with deferred=True - Persists the parameters to shared result storage - Returns a PrefectDistributedFuture with a task_run_id - A Task Worker process (that you run) pulls and executes the work - You can poll state, fetch results, and read logs from your FastAPI app Where it runs - In a separate Task Worker process you run with

serve(my_task)

— not in your FastAPI process/thread What you get back - A PrefectDistributedFuture that exposes: - `task_run_id`: UUID for tracking -

.state

/

.wait()

/

.result()

for status and results How to implement with FastAPI 1) Define and serve the task - Run this worker as a separate process/container. Ensure it shares result storage with your API (e.g., same mounted volume or remote storage like S3).

from prefect import task
from prefect.task_worker import serve

@task(log_prints=True)
def long_job(payload: dict) -> dict:
    import time
    print("Starting job")
    time.sleep(10)
    return {"ok": True, "size": len(payload)}

if __name__ == "__main__":
    # Concurrency optional via limit=
    serve(long_job)

2) Submit from FastAPI and poll status/results

from fastapi import FastAPI
from uuid import UUID
from prefect.client.orchestration import get_client
from app.tasks import long_job  # import the task above

app = FastAPI()

@app.post("/jobs", status_code=202)
async def start_job(payload: dict):
    f = long_job.delay(payload)
    return {"task_run_id": str(f.task_run_id)}

@app.get("/jobs/{task_run_id}")
async def get_job(task_run_id: UUID):
    async with get_client() as client:
        tr = await client.read_task_run(task_run_id)
        state = tr.state

        resp = {
            "id": str(task_run_id),
            "state": (state.type.value if state else "PENDING"),
            "name": (state.name if state else None),
            "message": (state.message if state else None),
        }

        if state and state.is_completed():
            try:
                resp["result"] = state.result(_sync=True)
            except Exception as e:
                resp["result_error"] = str(e)
        elif state and state.is_failed():
            try:
                resp["error"] = str(state.result(_sync=True))
            except Exception:
                resp["error"] = state.message

        return resp

3) Fetch logs for a task run - Logs are visible in the Prefect UI and also programmatically via the client. Make sure you use

get_run_logger()

or set

log_prints=True

to capture prints.

from uuid import UUID
from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FilterSet, LogFilter, LogFilterTaskRunId

@app.get("/jobs/{task_run_id}/logs")
async def get_logs(task_run_id: UUID, limit: int = 200, offset: int = 0):
    async with get_client() as client:
        filters = FilterSet(
            logs=LogFilter(task_run_id=LogFilterTaskRunId(any_=[task_run_id]))
        )
        logs = await client.read_logs(filter=filters, limit=limit, offset=offset)

    return [
        {
            "timestamp": str(log.timestamp),
            "level": log.level,
            "message": log.message,
        }
        for log in logs
    ]

When to choose a flow instead - Use a flow (and typically a deployment) when you have: - Multiple steps/tasks with dependencies

- Schedules, SLAs, retries across tasks - Infrastructure execution (e.g., Kubernetes/Docker workers for flows) - From FastAPI, you’d trigger the flow using

run_deployment(...)

and track the returned flow_run_id; you can then inspect both the flow run and its underlying task runs. - See run_deployment API Notes for production - Ensure your API and Task Worker share result storage so results/parameters can be retrieved: - For containers, mount the same

PREFECT_LOCAL_STORAGE_PATH

- Or configure remote result storage (e.g., S3/GCS) - Control worker concurrency via

serve(..., limit=N)

- Prefer

get_run_logger()

inside tasks for structured logs Docs and references - Background tasks quickstart: Run background tasks - Full FastAPI + background tasks example: Deploy a web app powered by background tasks - Task Worker API reference: prefect.task_worker - Futures API: prefect.futures - Client orchestration API: prefect.client.orchestration - REST logs endpoint: Read logs Recommendation for your case - For “invoke long-running logic from FastAPI and query state/logs/result,” use

task.delay()

+ a Task Worker. It’s the simplest, most direct pattern for API-driven background jobs with full observability. If you share a bit about your deployment environment (single container vs multiple, Cloud vs local, where you want to store results), I can tailor the storage and worker setup exactly to your needs.

Short answer - If you need parallelism, retries per entity, and rich observability per unit of work, use a flow that fans out to tasks (flow + tasks with map/submit). - If you’re okay processing sequentially (or you’ll shard externally) and want the simplest FastAPI integration, a single background task via task.delay() that iterates and emits progress works well. - In both cases, use Prefect artifacts to report live progress and final counts. How I’d choose - Flow + tasks (fan-out) is best when: - You want to process the 200 entities concurrently - You want per-entity retries/failures without stopping the whole run - You want per-entity logs/states in the UI - Single background task is best when: - Work must be strictly sequential, or concurrency is handled elsewhere - You want minimal moving parts: one task worker, one task - You’ll emit progress yourself (artifacts) and return an aggregate result Tracking progress and counts - Use Prefect progress artifacts for a live percentage - Use a table and/or markdown artifact for a summary - For per-entity outcomes in the fan-out pattern, aggregate futures’ states in the flow and emit a summary artifact Pattern A: Flow + mapped task (recommended for parallelism)

from prefect import flow, task
from prefect.futures import wait
from prefect.artifacts import create_progress_artifact, update_progress_artifact, create_table_artifact
from prefect.task_runners import ThreadPoolTaskRunner

@task(retries=2, retry_delay_seconds=2, log_prints=True)
def process_entity(e):
    # your chained API calls here
    return {"id": e["id"], "status": "success"}

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def process_all(entities: list[dict]):
    total = len(entities)
    progress_id = create_progress_artifact(
        progress=0.0,
        key="batch-progress",
        description=f"Processing {total} entities"
    )
    futures = process_entity.map(entities)

    # As they complete, compute counts
    done, _ = wait(futures)
    successes = [f for f in done if f.state.is_completed()]
    failures = [f for f in done if f.state.is_failed()]

    update_progress_artifact(
        artifact_id=progress_id,
        progress=100.0,
        description=f"Completed: {len(successes)} success, {len(failures)} failed"
    )

    # Optional: detailed results table
    create_table_artifact(
        table=[{"id": f.result()["id"], "status": "success"} for f in successes]
              + [{"id": entities[i].get("id"), "status": "failed"} for i, f in enumerate(done) if f.state.is_failed()],
        key="batch-results",
        description="Per-entity outcomes"
    )

    return {
        "total": total,
        "successful": len(successes),
        "failed": len(failures),
    }

Pattern B: Single background task that iterates and emits progress (simplest FastAPI integration) ``` from prefect import task from prefect.task_worker import serve from prefect.artifacts import create_progress_artifact, update_progress_artifact, create_table_artifact @task(log_prints=True) def process_batch(entities: list[dict], job_id: str): total = len(entities) progress_id = create_progress_artifact( progress=0.0, key=f"job-{job_id}-progress", description=f"Starting job {job_id} with {total} entities" ) successes, failures = [], [] for i, e in enumerate(entities, start=1): try: # your chained API calls here successes.append({"id": e["id"], "status": "success"}) except Exception as exc: failures.append({"id": e["id"], "status": "failed", "error": str(exc)}) update_progress_artifact( artifact_id=progress_id, progress=(i / total) * 100.0, description=f"{i}/{total} processed | {len(successes)} success, {len(failures)} failed" ) create_table_artifact( table=successes + failures, key=f"job-{job_id}-results",

description=f"Results for {job_id}" ) return { "job_id": job_id, "total": total, "successful": len(successes), "failed": len(failures), } if name == "__main__": serve(process_batch)

FastAPI endpoints for submission and status
- With Pattern B, submit via `process_batch.delay(...)` and track with the returned task_run_id. Poll state and artifacts for progress and final counts.
- With Pattern A, you can either call the flow directly from the API (if acceptable) or deploy it and trigger with `run_deployment`, then track by flow_run_id.

Helpful client snippets
- Read task run or flow run state/result in your API:

from uuid import UUID from prefect.client.orchestration import get_client async def read_task_status(task_run_id: UUID): async with get_client() as client: tr = await client.read_task_run(task_run_id) st = tr.state if not st: return {"state": "PENDING"} if st.is_completed(): return {"state": "COMPLETED", "result": st.result(_sync=True)} if st.is_failed(): return {"state": "FAILED", "message": st.message} return {"state": st.type.value}

- Read logs for a given task run:

from prefect.client.schemas.filters import FilterSet, LogFilter, LogFilterTaskRunId async def read_task_logs(task_run_id: UUID, limit=200, offset=0): async with get_client() as client: fs = FilterSet(logs=LogFilter(task_run_id=LogFilterTaskRunId(any_=[task_run_id]))) return await client.read_logs(filter=fs, limit=limit, offset=offset) ``` Docs to dive deeper - Run work concurrently (map/submit): Run work concurrently - Background tasks with FastAPI and artifacts: Deploy a web app powered by background tasks - Progress artifacts API: prefect.artifacts - Client orchestration API: prefect.client.orchestration If you share your constraints (need for parallelism, desired throughput, infra limits), I can recommend a concrete pattern and wire up the exact FastAPI + Prefect code you’ll need.

You’re right: if you only update progress at the start and end of a mapped flow, you’ll see 0 → 100. The fix is to update progress incrementally from the flow as each mapped task completes. Best practice for Pattern A: update a ProgressArtifact as futures complete - Use as_completed to iterate over mapped task futures as they finish. - Keep running counts of completed/success/failed. - Update a single ProgressArtifact each time one finishes. - Give the artifact a stable key (e.g., based on flow_run.id) so your frontend can poll it. Example

from prefect import flow, task
from prefect.futures import as_completed
from prefect.artifacts import create_progress_artifact, update_progress_artifact
from prefect.task_runners import ThreadPoolTaskRunner
from prefect.runtime import flow_run

@task(retries=2, retry_delay_seconds=2, log_prints=True)
def process_entity(e):
    # your chained API calls here
    return {"id": e["id"], "status": "success"}

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def process_all(entities: list[dict]):
    total = len(entities)
    prog_key = f"batch-{flow_run.id}-progress"

    # Create a single progress artifact we update as tasks finish
    prog_id = create_progress_artifact(
        progress=0.0,
        key=prog_key,
        description=f"Processing {total} entities"
    )

    # Fan out
    futures = process_entity.map(entities)

    # Incremental progress as futures complete
    completed = 0
    success = 0
    failed = 0

    for fut in as_completed(futures):
        completed += 1
        if fut.state.is_completed():
            success += 1
        else:
            failed += 1

        pct = (completed / total) * 100.0
        update_progress_artifact(
            artifact_id=prog_id,
            progress=pct,
            description=f"{completed}/{total} processed | {success} succeeded, {failed} failed"
        )

    # Optionally return final counts
    return {"total": total, "successful": success, "failed": failed, "progress_artifact_key": prog_key}

Notes - This keeps progress accurate without each task needing to know the total or fight over updates. - Your frontend can poll the artifact by its key (prog_key) to render a progress bar and counts. - If you want more frequent updates (e.g., partial stages inside each entity), you can also create per-entity artifacts or push intermediate logs, but the above is the simplest aggregate progress. If you prefer a single background task (Pattern B), you can still increment progress for every entity inside that task; but you’ll lose per-entity fault isolation and parallelism. Helpful docs - Futures (as_completed): prefect.futures - Progress artifacts: prefect.artifacts If you want, share how you’ll poll progress (Prefect API vs your own DB) and I’ll show the exact read_artifacts filter to match your keys and return counts efficiently.

In Pattern A it’s up to how you query. Prefect records logs with both flow_run_id and (for task logs) task_run_id. So: - If you query logs by flow_run_id, you’ll get a merged timeline: flow-level logs plus all task logs for that flow. - If you query by task_run_id, you’ll get only that specific task’s logs. - If you want “flow-only” logs (excluding tasks), query by flow_run_id and filter client-side for records where task_run_id is null/absent. Typical frontend queries - All logs for one flow run (merged flow + tasks):

from prefect.client.orchestration import get_client
from prefect.client.schemas.filters import FilterSet, LogFilter, LogFilterFlowRunId

async with get_client() as client:
    filters = FilterSet(
        logs=LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))
    )
    logs = await client.read_logs(filter=filters, limit=1000, offset=0)

- Only one task’s logs:

from prefect.client.schemas.filters import FilterSet, LogFilter, LogFilterTaskRunId

filters = FilterSet(
    logs=LogFilter(task_run_id=LogFilterTaskRunId(any_=[task_run_id]))
)
logs = await client.read_logs(filter=filters, limit=1000, offset=0)

Notes - Task logs include both flow_run_id and task_run_id; flow-only logs include only flow_run_id. - To see prints in logs, set

log_prints=True

on your tasks/flow or use

get_run_logger()

. - For background tasks (task.delay without a flow), logs will have a task_run_id but no flow_run_id. Docs - Read logs via API: REST: read logs - Client orchestration API: Python client

Short answer: don’t pass a live httpx.Client between tasks. Why - Tasks may run in different threads, processes, or even different machines (e.g., with `.delay()`/workers). A live client isn’t serializable and won’t reliably cross those boundaries. - Even within a threaded runner, sharing a mutable cookie jar across tasks is fragile and hard to coordinate when refreshes happen. Robust patterns that work well with Prefect 1) Recreate clients per task from a shared cookie store + a refresh lock - Store the current session cookies in a central place (Redis/DB/S3/etc.). - Each task builds its own

httpx.Client

from the stored cookies. - On 401, acquire a distributed lock so only one task refreshes the session and writes new cookies. Other tasks wait, then retry with the new cookies. Sketch

# Pseudocode – use your own Redis/DB client

SESSION_KEY = "api_session_cookies"
LOCK_KEY = "api_session_refresh_lock"

def load_cookies():
    raw = redis.get(SESSION_KEY)
    return httpx.Cookies() if not raw else httpx.Cookies( cookies=json.loads(raw) )

def save_cookies(cookies: httpx.Cookies):
    redis.set(SESSION_KEY, json.dumps(list(cookies.jar)), ex=3600)

def build_client_from_store() -> httpx.Client:
    cookies = load_cookies()
    return httpx.Client(base_url=BASE_URL, cookies=cookies, timeout=30)

def perform_login() -> httpx.Client:
    c = httpx.Client(base_url=BASE_URL, timeout=30)
    # ... do login, set cookies ...
    return c

@task(retries=0)
def process_entity(e):
    def do_call(client):
        return client.get(f"/entities/{e['id']}")

    client = build_client_from_store()
    r = do_call(client)
    if r.status_code != 401:
        return r.json()

    # 401 → attempt coordinated refresh
    with redis.lock(LOCK_KEY, timeout=60):  # ensure only one task refreshes
        # Re-check in case another task already refreshed
        client = build_client_from_store()
        r = do_call(client)
        if r.status_code == 401:
            # Still unauthorized → we refresh
            client = perform_login()
            save_cookies(client.cookies)

    # Retry original call with refreshed cookies
    client = build_client_from_store()
    r = do_call(client)
    r.raise_for_status()
    return r.json()

Notes - Represent cookies in a portable format (e.g., a dict/list or a “Cookie” header string) when storing. - Use a distributed lock (e.g., Redis lock). Prefect’s built-in memory/filesystem locks won’t help across machines. 2) Chunked fan‑out to reduce contention - Split the 200 entities into N chunks. - Each task processes its chunk with its own client/session and manages its own refresh on 401. - You still get parallelism, but you don’t need global cookie sharing. 3) Single owner of the session inside one long-running background task - Run Pattern B (a single background task via

.delay()

), create one

httpx.Client

, and iterate all entities inside that task. - You can still do internal concurrency with

httpx.AsyncClient

+ a semaphore, but it’s all within the same task/process, so one session is easy to manage. - Trade-off: you lose per-entity task states. What not to do - Don’t pass

httpx.Client

as a task parameter or rely on global variables for it; it will break when you change runners or introduce distribution. If you share how you’re running (single machine vs multiple workers, desired parallelism), I can tailor one of the patterns with concrete code (e.g., Redis-backed cookie store + lock, or an async single-owner task).

Great — single machine, 12 cores, Pattern 1 with Redis-backed cookies and a lock, and a hard cap of 10 concurrent entities. Below is a complete, working template. What it gives you - Each task builds its own httpx.Client from cookies stored in Redis - On 401, tasks coordinate a single refresh with a Redis lock, then retry - Flow fans out to tasks with at most 10 concurrent workers - A single ProgressArtifact is updated incrementally as tasks finish Requirements -

pip install prefect httpx redis

- Environment:

REDIS_URL

,

API_BASE_URL

,

API_USERNAME

,

API_PASSWORD

Code ``` import os import json import httpx import redis from typing import Any from prefect import flow, task, get_run_logger from prefect.futures import as_completed from prefect.task_runners import ThreadPoolTaskRunner from prefect.artifacts import create_progress_artifact, update_progress_artifact from prefect.runtime import flow_run # ------------------------------------------------------------------------------ # Configuration # ------------------------------------------------------------------------------ REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com") LOGIN_USERNAME = os.getenv("API_USERNAME") LOGIN_PASSWORD = os.getenv("API_PASSWORD") SESSION_KEY = "api_session_cookies" # Redis key to store cookies JSON LOCK_KEY = "api_session_refresh_lock" # Redis lock key for session refresh LOCK_TIMEOUT = 60 # Seconds the lock is held LOCK_BLOCKING_TIMEOUT = 30 # Seconds tasks will wait to acquire lock # ------------------------------------------------------------------------------ # Redis helpers # ------------------------------------------------------------------------------ def get_redis() -> redis.Redis: # decode_responses=True so we get strings back and can json.dumps/loads easily return redis.Redis.from_url(REDIS_URL, decode_responses=True) # ------------------------------------------------------------------------------ # Cookie (de)serialization helpers # ------------------------------------------------------------------------------ def serialize_cookies(cookies: httpx.Cookies) -> str: # Convert CookieJar to a JSON-serializable list of dicts data = [] for c in cookies.jar: data.append( { "name": c.name, "value": c.value, "domain": c.domain, "path": c.path, "expires": c.expires, "secure": bool(c.secure), } ) return json.dumps(data) def deserialize_cookies(raw: str | None) -> httpx.Cookies: jar = httpx.Cookies() if not raw: return jar for c in json.loads(raw): # domain/path are optional for client-side cookies; include if present jar.set( c["name"], c["value"], domain=c.get("domain"), path=c.get("path"), ) return jar def load_cookies() -> httpx.Cookies: r = get_redis() raw = r.get(SESSION_KEY) return deserialize_cookies(raw) def save_cookies(cookies: httpx.Cookies, ttl_seconds: int | None = None) -> None: r = get_redis() payload = serialize_cookies(cookies) if ttl_seconds: r.setex(SESSION_KEY, ttl_seconds, payload) else: r.set(SESSION_KEY, payload) # ------------------------------------------------------------------------------ # HTTP client + login # ------------------------------------------------------------------------------ def build_client_from_store() -> httpx.Client: return httpx.Client(base_url=BASE_URL, timeout=30, cookies=load_cookies()) def login_and_get_client() -> httpx.Client: """ Perform a fresh login and persist the resulting cookies to Redis. Replace the login endpoint/payload with your real auth flow. """ if not LOGIN_USERNAME or not LOGIN_PASSWORD:

raise RuntimeError("Missing API_USERNAME/API_PASSWORD environment variables") client = httpx.Client(base_url=BASE_URL, timeout=30) # Example login flow resp = client.post( "/auth/login", json={"username": LOGIN_USERNAME, "password": LOGIN_PASSWORD}, ) resp.raise_for_status() save_cookies(client.cookies) return client # ------------------------------------------------------------------------------ # Your chained API calls per entity # ------------------------------------------------------------------------------ def process_entity_chain(client: httpx.Client, entity: dict[str, Any]) -> dict[str, Any]: """ Example chained calls — replace with your real logic. """ # GET (may require auth via cookies) r1 = client.get(f"/entities/{entity['id']}") r1.raise_for_status() # POST (work step) r2 = client.post(f"/entities/{entity['id']}/process", json={"payload": entity}) r2.raise_for_status() return { "id": entity["id"], "status": "success", "result": r2.json(), } # ------------------------------------------------------------------------------ # Task with 401 handling + coordinated session refresh via Redis lock # ------------------------------------------------------------------------------ @task(retries=0, log_prints=True) def process_entity(entity: dict[str, Any]) -> dict[str, Any]: logger = get_run_logger() def try_chain_with(client: httpx.Client) -> dict[str, Any]: return process_entity_chain(client, entity) # First attempt with existing cookies client = build_client_from_store() try: return try_chain_with(client) except httpx.HTTPStatusError as e: if e.response.status_code != 401: # Not an auth issue — bubble up raise # 401 unauthorized: coordinate a refresh rds = get_redis() lock = rds.lock(LOCK_KEY, timeout=LOCK_TIMEOUT, blocking_timeout=LOCK_BLOCKING_TIMEOUT) acquired = lock.acquire(blocking=True) try: # Re-check with latest cookies in case another task refreshed while we waited client = build_client_from_store() try: return try_chain_with(client) except httpx.HTTPStatusError as e2: if e2.response.status_code != 401: raise # Still unauthorized → we do the refresh logger.info("Refreshing session cookies after 401") client = login_and_get_client() # Final retry return try_chain_with(client) finally: if acquired: try: lock.release() except Exception: # Lock may already be released or expired — best effort pass # ------------------------------------------------------------------------------ # Flow with max concurrency = 10 and incremental progress # ------------------------------------------------------------------------------ @flow(task_runner=ThreadPoolTaskRunner(max_workers=10)) def process_all_entities(entities: list[dict[str, Any]]) -> dict[str, Any]: total = len(entities) prog_key = f"batch-{flow_run.id}-progress" prog_id = create_progress_artifact( progress=0.0, key=prog_key, description=f"Processing {total} entities" ) futures = process_entity.map(entities) completed = success = failed = 0 for fut in as_completed(futures): completed += 1 try: _ = fut.result() success += 1 except Exception: failed += 1 pct = (completed / total) * 100.0 if total else 100.0 update_progress_artifact( artifact_id=prog_id, progress=pct, description=f"{completed}/{total} processed | {success} success, {failed} failed" ) return { "total": total, "successful": success, "failed": failed,

"progress_artifact_key": prog_key, } # ------------------------------------------------------------------------------ # Optional: bootstrap a session once at the beginning (not required) # ------------------------------------------------------------------------------ @flow def ensure_session(): """ Optional: proactively login once to prime the cookie store. You can call this before process_all_entities if you like. """ try: login_and_get_client() except Exception as e: raise RuntimeError(f"Failed to establish session: {e}") ``` How to use - Call

process_all_entities(entities)

where

entities

is a list of dicts like

{"id": "123", ...}

. - Your frontend can poll the single progress artifact by key

batch-{flow_run_id}-progress

and show counts. - Concurrency is capped at 10 via

ThreadPoolTaskRunner(max_workers=10)

. Notes - Each task uses its own httpx.Client instance; no clients are shared across tasks. - Only one task performs a login refresh on 401, guarded by a Redis lock; others wait and then reuse the refreshed cookies. - You can adjust `LOCK_TIMEOUT`/`LOCK_BLOCKING_TIMEOUT` based on how long a login may take. Docs - Run work concurrently (map/as_completed): Run work concurrently - Artifacts (progress, tables, markdown): prefect.artifacts If you want to add a pre-check to keep the session fresh before a burst (e.g.,

ensure_session()

), or switch to async with

httpx.AsyncClient

, I can adapt the template accordingly.

Awesome — here’s a complete async version using httpx.AsyncClient, Redis (async), a Redis-backed refresh lock, and a hard cap of 10 concurrent entity tasks using Prefect’s concurrency manager. Important setup - Install: pip install prefect httpx redis - Environment variables: REDIS_URL, API_BASE_URL, API_USERNAME, API_PASSWORD - Create a Prefect concurrency limit named entity-processing with limit = 10 in the UI (Settings -> Concurrency limits). This enforces the cap across all running tasks. - Then we’ll acquire slots in each task with

async with prefect.concurrency.asyncio.concurrency("entity-processing")

. Code ``` import os import json import httpx import redis.asyncio as aioredis from typing import Any from prefect import flow, task, get_run_logger from prefect.futures import as_completed from prefect.task_runners import PrefectTaskRunner from prefect.artifacts import acreate_progress_artifact, aupdate_progress_artifact from prefect.runtime import flow_run from prefect.concurrency.asyncio import concurrency # requires a configured concurrency limit # ------------------------------------------------------------------------------ # Configuration # ------------------------------------------------------------------------------ REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0") BASE_URL = os.getenv("API_BASE_URL", "https://api.example.com") LOGIN_USERNAME = os.getenv("API_USERNAME") LOGIN_PASSWORD = os.getenv("API_PASSWORD") SESSION_KEY = "api_session_cookies" # Redis key to store cookies JSON LOCK_KEY = "api_session_refresh_lock" # Redis lock key for session refresh LOCK_TIMEOUT = 60 # seconds the lock is held server-side LOCK_BLOCKING_TIMEOUT = 30 # seconds tasks will wait to acquire lock CONCURRENCY_LIMIT_NAME = "entity-processing" # define this in Prefect UI, limit=10 # ------------------------------------------------------------------------------ # Redis helpers (async) # ------------------------------------------------------------------------------ def get_redis() -> aioredis.Redis: # decode_responses=True to read/write strings return aioredis.from_url(REDIS_URL, decode_responses=True) # ------------------------------------------------------------------------------ # Cookie (de)serialization helpers # ------------------------------------------------------------------------------ def serialize_cookies(cookies: httpx.Cookies) -> str: data = [] for c in cookies.jar: data.append( { "name": c.name, "value": c.value, "domain": c.domain, "path": c.path, "expires": c.expires, "secure": bool(c.secure), } ) return json.dumps(data) def deserialize_cookies(raw: str | None) -> httpx.Cookies: jar = httpx.Cookies() if not raw: return jar for c in json.loads(raw): jar.set(c["name"], c["value"], domain=c.get("domain"), path=c.get("path")) return jar async def load_cookies() -> httpx.Cookies: r = get_redis() raw = await r.get(SESSION_KEY) return deserialize_cookies(raw) async def save_cookies(cookies: httpx.Cookies, ttl_seconds: int | None = None) -> None: r = get_redis() payload = serialize_cookies(cookies) if ttl_seconds: await r.setex(SESSION_KEY, ttl_seconds, payload) else: await r.set(SESSION_KEY, payload) # ------------------------------------------------------------------------------ # HTTP client + login (async) # ------------------------------------------------------------------------------ async def build_client_from_store() -> httpx.AsyncClient: return httpx.AsyncClient(base_url=BASE_URL, timeout=30, cookies=await load_cookies()) async def login_and_get_client() -> httpx.AsyncClient: """ Perform a fresh login and persist cookies to Redis.

Replace the endpoint/payload with your real auth flow. """ if not LOGIN_USERNAME or not LOGIN_PASSWORD: raise RuntimeError("Missing API_USERNAME/API_PASSWORD environment variables") client = httpx.AsyncClient(base_url=BASE_URL, timeout=30) resp = await client.post( "/auth/login", json={"username": LOGIN_USERNAME, "password": LOGIN_PASSWORD}, ) resp.raise_for_status() await save_cookies(client.cookies) return client # ------------------------------------------------------------------------------ # Your chained API calls per entity (async) # ------------------------------------------------------------------------------ async def process_entity_chain(client: httpx.AsyncClient, entity: dict[str, Any]) -> dict[str, Any]: """ Example chained calls — replace with your real logic. """ r1 = await client.get(f"/entities/{entity['id']}") r1.raise_for_status() r2 = await client.post(f"/entities/{entity['id']}/process", json={"payload": entity}) r2.raise_for_status() return { "id": entity["id"], "status": "success", "result": r2.json(), } # ------------------------------------------------------------------------------ # Task with (1) Prefect concurrency cap + (2) Redis-coordinated session refresh # ------------------------------------------------------------------------------ @task(retries=0, log_prints=True) async def process_entity(entity: dict[str, Any]) -> dict[str, Any]: """ - Ensures at most N tasks enter concurrently via Prefect concurrency ("entity-processing"). - If any call returns 401, one task refreshes cookies under a Redis lock; others wait and retry. """ logger = get_run_logger() async with concurrency(CONCURRENCY_LIMIT_NAME): # attempt with current cookies client = await build_client_from_store() try: return await process_entity_chain(client, entity) except httpx.HTTPStatusError as e: if e.response.status_code != 401: raise # 401 unauthorized: coordinate a refresh r = get_redis() lock = r.lock(LOCK_KEY, timeout=LOCK_TIMEOUT, blocking_timeout=LOCK_BLOCKING_TIMEOUT) acquired = await lock.acquire(blocking=True) try: # Re-check in case someone else refreshed client = await build_client_from_store() try: return await process_entity_chain(client, entity) except httpx.HTTPStatusError as e2: if e2.response.status_code != 401: raise logger.info("Refreshing session cookies after 401") client = await login_and_get_client() # Final retry return await process_entity_chain(client, entity) finally: if acquired: try: await lock.release() except Exception: pass # ------------------------------------------------------------------------------ # Flow with incremental progress (async) and PrefectTaskRunner for async tasks # ------------------------------------------------------------------------------ @flow(task_runner=PrefectTaskRunner()) async def process_all_entities(entities: list[dict[str, Any]]) -> dict[str, Any]: total = len(entities) prog_key = f"batch-{flow_run.id}-progress" prog_id = await acreate_progress_artifact( progress=0.0, key=prog_key, description=f"Processing {total} entities" ) futures = process_entity.map(entities) completed = success = failed = 0 # Iterate as each task finishes; these futures are done when yielded for fut in as_completed(futures): completed += 1 try: _ = fut.result() # safe because it's already finished success += 1 except Exception: failed += 1

pct = (completed / total) * 100.0 if total else 100.0 await aupdate_progress_artifact( artifact_id=prog_id, progress=pct, description=f"{completed}/{total} processed | {success} success, {failed} failed" ) return { "total": total, "successful": success, "failed": failed, "progress_artifact_key": prog_key, } # ------------------------------------------------------------------------------ # Optional: proactively establish a session # ------------------------------------------------------------------------------ @flow(task_runner=PrefectTaskRunner()) async def ensure_session(): try: await login_and_get_client() except Exception as e: raise RuntimeError(f"Failed to establish session: {e}") ``` How concurrency is enforced - The Prefect-side cap of 10 is enforced by the

async with concurrency("entity-processing")

manager. Be sure you’ve created a concurrency limit named entity-processing with a value of 10 in the Prefect UI. - This approach works regardless of task runner and guarantees no more than 10 concurrent tasks will execute the body at the same time. Why not use a shared AsyncClient? - Prefect tasks can run in different threads/contexts; sharing a live client is brittle. Rebuilding a client from a shared cookie store keeps things safe and deterministic, and the Redis lock ensures only one refresh occurs on 401s. Related docs - Run work concurrently: Run work concurrently - Artifacts (async functions like acreate_progress_artifact): prefect.artifacts If you want me to add a minimal endpoint that triggers this flow and a progress endpoint that polls the progress artifact by key, I can drop that in too.

Great — since you already have multiple flows and tasks and your API runs in Docker on a single 12‑core machine, you have two solid options for “serving” your flows so your FastAPI endpoint can trigger them: Option 1: Serve flows locally with a Runner (simple, single machine) - Use flow.to_deployment(...) and serve(...) to create deployments and start a Runner process that executes flow runs in subprocesses on the same machine. - Your FastAPI endpoint then calls run_deployment(...) to trigger a run and returns a flow_run_id for polling. - Set limit to control concurrent flow-run processes (e.g., 8 on a 12‑core box). Serve script (run alongside your API)

# serve_flows.py
from prefect import flow, serve

# import your flows here
from my_project.flows import flow_a, flow_b, flow_c

if __name__ == "__main__":
    dep_a = flow_a.to_deployment(name="flow-a")
    dep_b = flow_b.to_deployment(name="flow-b")
    dep_c = flow_c.to_deployment(name="flow-c")

    # Limit = max concurrent flow run subprocesses on this machine
    serve(dep_a, dep_b, dep_c, limit=8, pause_on_shutdown=False)

FastAPI endpoint to trigger a flow

# app.py
from fastapi import FastAPI, HTTPException
from prefect.deployments import run_deployment
from prefect.client.orchestration import get_client
from uuid import UUID

app = FastAPI()

@app.post("/flows/flow-a/trigger")
async def trigger_flow_a(payload: dict):
    try:
        fr = await run_deployment("flow_a/flow-a", parameters=payload, timeout=0)
        return {"flow_run_id": str(fr.id)}
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/flows/status/{flow_run_id}")
async def flow_status(flow_run_id: str):
    async with get_client() as client:
        fr = await client.read_flow_run(UUID(flow_run_id))
        state = fr.state
        return {
            "flow_run_id": str(fr.id),
            "state": state.type if state else None,
            "name": state.name if state else None,
            "message": state.message if state else None,
            "completed": state.is_completed() if state else False,
        }

Containerization notes - Run the serve script and your API in the same container, or separate containers, both pointed at the same Prefect API. - Example CMD to run both in one container:

CMD python serve_flows.py & uvicorn app:app --host 0.0.0.0 --port 8000

- Set PREFECT_API_URL in the container env to your Prefect Server/Cloud. When to pick this - You’re on one machine, prefer simplicity, and don’t need dynamic infra. Each flow run is isolated in its own subprocess; limit=8 is a good starting point for a 12‑core host. Option 2: Deploy to a work pool and run a worker (more scalable) - Use flow.deploy(..., work_pool_name="...") to register deployments that run on a worker you start. - Start a worker:

prefect worker start --pool my-process-pool

- Your FastAPI endpoint still uses run_deployment(...). This makes it easier to scale later or move to Docker/K8s workers, but it’s more setup than serve(). Which should you use? - For your current setup (single host, Dockerized API), Option 1 with serve(...) is the quickest and most reliable. You’ll get: - Non-blocking trigger from your API via run_deployment - Observability in the UI - Concurrency control via serve(limit=...) - Subprocess isolation for each flow run Tips - Align serve(limit=...) with your workload. If flows are CPU-heavy, try 6–8. If mostly I/O-bound, you can go higher. - If your flows themselves fan out to tasks with map/submit, their own task_runner concurrency will further consume cores/threads. Monitor CPU and adjust. - Add tags on to_deployment for routing and filtering. - You can serve multiple flows together; they share the same limit across the single Runner. Docs - Serve flows locally: Serve flows

- Trigger with run_deployment: run_deployment - Deployments overview: Deployments If you paste your flow names and parameters, I’ll generate a tailored serve_flows.py and FastAPI endpoints for each.