https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Before creating a new worker how can ...
# ask-marvin
v
@Marvin Before creating a new worker how can I setup the namespace and image name for the prefect workpool through entrypoint.sh
m
✅ thought for 478.7 seconds
Great question. In Prefect 3.x, defaults like image and (for Kubernetes) namespace live in a work pool’s base job template. There aren’t CLI flags like `--image`/`--namespace` on `prefect work-pool create`; instead, you supply a JSON base job template when creating or updating the pool. That’s easy to automate from an entrypoint.sh. Below are examples you can drop into an entrypoint.sh to set these before starting your worker. Assumptions: - You’re targeting Kubernetes if you need a namespace; I’ve included Docker as well in case that’s useful. - You have 
jq
available in the container for JSON edits. Option A: Kubernetes work pool (set namespace and image defaults)
Copy code
#!/usr/bin/env bash
set -euo pipefail

# Inputs (env vars)
WORK_POOL_NAME="${WORK_POOL_NAME:-my-k8s-pool}"
WORK_POOL_TYPE="kubernetes"
NAMESPACE="${NAMESPACE:-default}"
IMAGE="${IMAGE:-my-registry/my-image:latest}"

TMP_TEMPLATE="$(mktemp)"
PATCHED_TEMPLATE="$(mktemp)"

# 1) Get the default Kubernetes base job template
prefect work-pool get-default-base-job-template --type "$WORK_POOL_TYPE" --file "$TMP_TEMPLATE"

# 2) Patch namespace and image
#    - Sets the pool-level default namespace
#    - Sets the container image in the Job manifest
jq --arg ns "$NAMESPACE" --arg img "$IMAGE" '
  .job_configuration.namespace = $ns
  | .job_configuration.job_manifest.metadata.namespace = $ns
  | .job_configuration.job_manifest.spec.template.spec.containers[0].image = $img
' "$TMP_TEMPLATE" > "$PATCHED_TEMPLATE"

# 3) Create or update the work pool with this template
prefect work-pool create "$WORK_POOL_NAME" \
  --type "$WORK_POOL_TYPE" \
  --base-job-template "$PATCHED_TEMPLATE" \
  --overwrite

# Optionally start the worker (uncomment if desired):
# prefect worker start -p "$WORK_POOL_NAME"
Option B: Docker work pool (set image default)
Copy code
#!/usr/bin/env bash
set -euo pipefail

WORK_POOL_NAME="${WORK_POOL_NAME:-my-docker-pool}"
WORK_POOL_TYPE="docker"
IMAGE="${IMAGE:-my-registry/my-image:latest}"

TMP_TEMPLATE="$(mktemp)"
PATCHED_TEMPLATE="$(mktemp)"

prefect work-pool get-default-base-job-template --type "$WORK_POOL_TYPE" --file "$TMP_TEMPLATE"

# For Docker workers the image is at job_configuration.image
jq --arg img "$IMAGE" '
  .job_configuration.image = $img
' "$TMP_TEMPLATE" > "$PATCHED_TEMPLATE"

prefect work-pool create "$WORK_POOL_NAME" \
  --type "$WORK_POOL_TYPE" \
  --base-job-template "$PATCHED_TEMPLATE" \
  --overwrite

# Optionally start the worker:
# prefect worker start -p "$WORK_POOL_NAME"
Notes and alternatives: - Instead of hard-coding the values, you can keep them overridable by using Jinja defaults in the template (e.g., set 
image
to 
{{ image | default('my-image:latest') }}
and 
namespace
to 
{{ namespace | default('my-namespace') }}
). You can do this by writing those strings into the JSON in step 2 if you prefer. - You can override these per-deployment using 
job_variables
in your deployments (e.g., in 
prefect.yaml
or with 
flow.deploy(..., job_variables={"image": "...", "namespace": "..."})
). - Inspect your pool to confirm the effective template: 
prefect work-pool inspect "$WORK_POOL_NAME"
Docs: - Work pool CLI: prefect work-pool CLI - Managing work pools: Manage work pools - Kubernetes worker guide: Kubernetes work pools - Overriding job variables: Customize job variables Quick check: are you targeting Kubernetes specifically? If so, I can tailor the template patch to include service account, resources, or other K8s fields you might want to default.
v
@Marvin this is flow definition async def alert_valiidator(): columns = await get_required_columns alerts = await get_configured_alerts base_query = await build_base_query alert_results = await validate_alerts This how I've defined it even though I defined them squentially why the flow run graph doesn't show like this?
m
✅ thought for 266.3 seconds
Short answer: the Prefect UI shows a dependency graph, not “the order you wrote the code.” If tasks don’t pass data to each other (or explicitly declare a dependency), the graph won’t show edges between them even if you awaited them sequentially. Common gotchas: - Make sure these are Prefect tasks (decorated with 
@task
) and that you’re actually calling them (e.g., 
await get_required_columns()
not 
await get_required_columns
). - Simply awaiting tasks one after another does not create graph edges; edges are created by data dependencies or explicit 
wait_for
. Two ways to make the graph show them sequentially: 1) Create real data dependencies (pass futures as inputs)
Copy code
from prefect import flow, task

@task
async def get_required_columns(): ...
@task
async def get_configured_alerts(columns): ...
@task
async def build_base_query(columns, alerts): ...
@task
async def validate_alerts(columns, alerts, base_query): ...

@flow
async def alert_validator():
    # Call tasks to get futures
    columns_f = get_required_columns()
    alerts_f = get_configured_alerts(columns=columns_f)
    base_query_f = build_base_query(columns=columns_f, alerts=alerts_f)
    result_f = validate_alerts(columns=columns_f, alerts=alerts_f, base_query=base_query_f)

    # Finally resolve the result
    return await result_f
Because each task takes upstream futures as inputs, Prefect infers edges: columns -> alerts -> base_query -> validate_alerts. 2) Use explicit ordering with wait_for (no data passed)
Copy code
from prefect import flow, task

@task
async def get_required_columns(): ...
@task
async def get_configured_alerts(): ...
@task
async def build_base_query(): ...
@task
async def validate_alerts(): ...

@flow
async def alert_validator():
    c = get_required_columns()
    a = get_configured_alerts(wait_for=[c])
    bq = build_base_query(wait_for=[a])
    v = validate_alerts(wait_for=[bq])
    return await v
Here 
wait_for
adds edges purely for ordering. Tips: - If you 
await
immediately and then call the next task with no reference to the prior task’s future, Prefect won’t draw a dependency edge. - Prefer passing futures (Option 1) when there’s a real data flow. Use 
wait_for
when you only need ordering. Docs: - Tasks and dependencies: Tasks concepts - Concurrency and dependencies (
wait_for
examples): Run work concurrently If you share how your tasks are decorated/called, I can tailor the snippet directly to your code.
v
@Marvin I have a flow which has couple of tasks for it. How can I handle exceptions in prefect flow and tasks ? • What would happen when I don't use try/except in the flows or tasks • If I have to properly handle exceptions how can I do it? what are the best practices were followed • Should I explicitly add try/except in every tasks and flow ? Or is there any way to do it handle exception gracefully ??
m
✅ thought for 345.0 seconds
Great question. Assuming you’re on Prefect 3.x, here’s how exceptions behave and the patterns we recommend. What happens if you don’t use try/except - Tasks: If a task raises an exception, Prefect catches it and marks the task Failed. The exception and traceback are attached to the task’s state and logged. - Flows: If the flow code raises an exception (and it’s not handled), the flow is marked Failed. - Downstream tasks: Any tasks that depend on a failed task are skipped by default (they won’t run), unless you explicitly allow the failure to propagate as data (see allow_failure below). How to handle exceptions properly (recommended patterns) - Use retries with backoff for transient errors - Set 
retries
and 
retry_delay_seconds
on tasks/flows. Use 
exponential_backoff
for backoff schedules. Optionally add 
retry_condition_fn
to only retry on certain errors. - Use timeouts to prevent hangs - Set 
timeout_seconds
on tasks/flows; when exceeded, the run fails (or retries if configured). - Use hooks for notifications/cleanup - 
on_failure
, 
on_completion
, etc. let you react when a task/flow changes state (e.g., send a Slack/Teams message, clean up resources). Hooks run after the state change and after all retries are exhausted for failures. - Make non-critical steps “soft-fail” with allow_failure - Wrap a task future in 
allow_failure(...)
to let downstream tasks run even if it failed. Downstream logic can inspect the state/exception and choose how to proceed. - Only use try/except inside a task when you can actually recover or want to convert an expected error to a controlled return value - If you swallow an unexpected exception, Prefect will think the task succeeded. Prefer letting Prefect mark it Failed (and retry) unless you have a specific recovery plan. Code examples 1) Retries, backoff, timeouts, and conditional retries
Copy code
from prefect import flow, task, get_run_logger, allow_failure
from prefect.tasks import exponential_backoff
from prefect.states import get_state_exception

def retry_if_network_error(state):
    exc = get_state_exception(state)
    # Adjust to your error types
    return isinstance(exc, (TimeoutError, ConnectionError))

@task(
    retries=5,
    retry_delay_seconds=exponential_backoff(2),  # 2, 4, 8, 16, ...
    retry_jitter_factor=0.1,                     # optional
    timeout_seconds=60,
    on_failure=[]                                # can add hooks here too
)
def fetch_data(url: str) -> str:
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Fetching {url}")
    # your code that may raise
    ...

@task
def parse_data(payload: str) -> dict:
    # will be skipped if fetch_data failed (unless you use allow_failure)
    ...

@flow(
    on_failure=[],           # flow-level hooks
    timeout_seconds=600,     # optional
)
def pipeline():
    f = fetch_data("<https://api.example.com/data>")
    result = parse_data(f)
    return result
To use conditional retries:
Copy code
@task(
    retries=5,
    retry_delay_seconds=exponential_backoff(2),
    retry_condition_fn=retry_if_network_error
)
def fetch_data(...):
    ...
2) Allowing a failure to continue downstream (soft-fail)
Copy code
from prefect import flow, task, allow_failure
from prefect.states import get_state_exception

@task
def optional_enrichment(x: int) -> int:
    # might raise
    ...

@task
def use_enrichment(x, enrichment):
    # If enrichment failed, enrichment state is available
    if enrichment.state and enrichment.state.is_failed():
        exc = get_state_exception(enrichment.state)
        # decide to proceed with defaults
        return {"x": x, "enriched": False, "reason": str(exc)}
    else:
        return {"x": x, "enriched": True, "value": enrichment.result()}

@flow
def my_flow(x: int):
    enrich_future = optional_enrichment(x)
    enrich_soft = allow_failure(enrich_future)   # let downstream run even if it failed
    return use_enrichment(x, enrich_soft)
Notes:
- 
allow_failure(...)
ensures downstream tasks are not skipped when an upstream task fails. It lets you inspect the state and decide how to proceed. - You can still call 
enrich_soft.result(raise_on_failure=False)
in downstream code if you want the value (or None) without raising, and then check 
enrich_soft.state
or 
get_state_exception(...)
. 3) Hooks for notifications or cleanup
Copy code
from prefect import flow, task, get_run_logger

def notify_task_failure(state):
    logger = get_run_logger()
    logger.error(f"Task failed: {state.name} -> {state.message}")

def notify_flow_failure(state):
    logger = get_run_logger()
    logger.error(f"Flow failed: {state.name} -> {state.message}")

@task(on_failure=[notify_task_failure])
def do_work():
    ...

@flow(on_failure=[notify_flow_failure])
def my_flow():
    do_work()
4) When to use try/except inside a task - Use it when you can recover or want to return a “safe” value with logging:
Copy code
@task
def read_maybe(s: str) -> str | None:
    logger = get_run_logger()
    try:
        return read_from_external(s)  # may raise
    except KnownParseError as e:
        logger.warning(f"Parse failed for {s}: {e}; returning None")
        return None
- Avoid broad try/except that hides unexpected errors; those should propagate so Prefect can mark the run Failed and retry if configured. Answers to your bullets - What happens without try/except? - The run fails (task/flow marked Failed), logs the exception, and downstream dependencies are skipped by default. - How to handle exceptions properly? - Prefer retries/backoff, timeouts, and hooks for notification/cleanup. Use allow_failure for non-critical steps. Use try/except only to recover from expected errors or to translate them into controlled results. - Do you need try/except everywhere? - No. Let Prefect manage failure states and retries. Only add try/except where you can meaningfully recover or want to convert known errors to safe outputs. Docs to dive deeper - Retries and backoff: Retries and backoff - State change hooks: State change hooks - Logging: Add logging - States and exceptions: States concept If you share a small snippet of your flow and tasks, I can suggest targeted patterns (where to add retries, allow_failure, or hooks).
v
@Marvin I want to stream flow run logs to Frontend when a flow starts it's execution i want start sending their longs to frontend via Streaming SSE how can I do that kinda of subscription or via prefect logger subscriber ?
m
✅ thought for 225.4 seconds
Yes—you can stream Prefect logs in real-time and pipe them to your frontend over SSE. In Prefect 3.x, the supported way is to subscribe to logs over a WebSocket using the logs subscriber, then bridge that stream to an SSE endpoint in your app. Key pieces: - Subscribe to logs with 
prefect.logging.clients.get_logs_subscriber(...)
- Filter by 
flow_run_id
so you only receive logs for that run - Expose an SSE endpoint in your backend that reads from the subscriber and yields SSE frames to the browser Here’s a complete, minimal FastAPI example that streams a single flow run’s logs over SSE:
Copy code
# Backend: FastAPI SSE endpoint that streams logs for a specific flow run
import json
import asyncio
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse

from prefect.logging.clients import get_logs_subscriber
from prefect.client.schemas.filters import LogFilter, LogFilterFlowRunId

app = FastAPI()

async def to_sse(log):
    payload = {
        "timestamp": log.timestamp.isoformat() if getattr(log, "timestamp", None) else None,
        "level": log.level,
        "message": log.message,
        "name": log.name,
        "flow_run_id": str(getattr(log, "flow_run_id", "") or ""),
        "task_run_id": str(getattr(log, "task_run_id", "") or ""),
    }
    return f"data: {json.dumps(payload)}\n\n"

@app.get("/streams/flows/{flow_run_id}/logs")
async def stream_flow_logs(flow_run_id: str, request: Request):
    # Only stream logs for this flow run
    log_filter = LogFilter(flow_run_id=LogFilterFlowRunId(any_=[flow_run_id]))

    # Auto-reconnect if the Prefect websocket drops
    subscriber = get_logs_subscriber(
        filter=log_filter,
        reconnection_attempts=20,
    )

    async def event_generator():
        try:
            async with subscriber as sub:
                async for log in sub:
                    # Stop if the browser disconnected
                    if await request.is_disconnected():
                        break
                    yield await to_sse(log)
        except asyncio.CancelledError:
            # Client disconnected
            pass

    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache"},
    )
Notes: - Prefect Cloud auth: set 
PREFECT_API_URL
and 
PREFECT_API_KEY
in the server environment where this runs. For OSS, set 
PREFECT_API_URL
to your server URL (e.g., http://127.0.0.1:4200/api). - This endpoint will start yielding events as soon as the run emits logs. You can call it as soon as the run starts (or earlier) and the stream will “wake up” when logs arrive. - For production, consider sending a periodic heartbeat like 
: ping\n\n
every 15–30s to keep proxies from closing idle SSE connections. If you want to automatically start streaming when any flow starts - You can subscribe to Prefect events for flow run state changes (e.g., “Running”), then attach log subscribers dynamically. This is useful for “global” feeds or dashboards. - Example outline: - Use events subscriber (from 
prefect.events.clients.get_events_subscriber
) filtered to 
prefect.flow-run.Running
- For each event, read the 
flow_run_id
and begin forwarding logs for that run to connected frontend clients - API ref for events clients: Events clients Optional: backfill historical logs on connect - You can fetch existing logs first (REST) then continue streaming new logs via the subscriber so your UI shows the full history: - REST read logs endpoint: Read logs (REST) - Then switch to streaming with the code above Relevant docs - Logs subscriber API: prefect.logging.clients
- Filters API: Filters (LogFilter, LogFilterFlowRunId) - Events clients: Events clients Does your setup run on Prefect Cloud or OSS? And do you want per-flow-run SSE endpoints (one stream per run) or a single global feed for “whatever is running now”? I can tailor the snippet accordingly.
v
@Marvin We are using prefect to sync airbyte data into our system.. it can run on schedule basis and can be run on on demand for on demand I can pass the job environment variable via run deployment and for deployment I can able to create it with job env variable while creating deployment however if the env changes later on deman run deployment might work right ?? but tehe already created deployment might not work since it was created with old env variables right ? how to mitigate this problem with prefect ?
m
✅ thought for 371.5 seconds
Great question — you’re right to think about this up front. Short answer - A deployment’s job variables are captured at creation time. If you hard-code env in the deployment, changing it later won’t be picked up unless you redeploy (or override per-run). - To avoid redeploys when env changes, don’t put changing values directly into the deployment’s job variables. Instead, use one of the patterns below. Recommended patterns to mitigate env drift 1) Centralize env in the Work Pool (preferred for infrastructure-level env) - Put env in your Work Pool’s default job variables (or base job template), and avoid overriding them at the deployment level. - Then, when env changes, update the Work Pool once — all deployments that don’t override those keys will pick up the new values for future runs automatically. - You can edit this in the UI (Work Pool > Default job variables) or via CLI by updating the base job template file:
Copy code
prefect work-pool update "<POOL_NAME>" --base-job-template ./job-template.json
CLI help: 
prefect work-pool update --help
2) Read config at runtime via Prefect Variables (preferred for app-level toggles/config) - Store mutable, non-sensitive configuration in Prefect Variables and read them in your flow code at run time. Update the variable — no redeploy needed. - Example:
Copy code
from prefect import flow
  from prefect.variables import Variable

  @flow
  def sync_airbyte():
      target_env = Variable.get("AIRBYTE_TARGET_ENV", default="prod")
      # use target_env in your logic
- Manage variables via CLI:
Copy code
prefect variable set AIRBYTE_TARGET_ENV staging
  prefect variable get AIRBYTE_TARGET_ENV
  prefect variable unset AIRBYTE_TARGET_ENV
Docs: Variables concept • CLI reference 3) One-off overrides at run time (on-demand runs) - For ad-hoc changes, override job variables per run: - CLI:
Copy code
prefect deployment run "<FLOW>/<DEPLOYMENT>" \
      --job-variable env='{"AIRBYTE_TARGET_ENV":"staging"}'
Note: the value is parsed as JSON; pass 
--job-variable
multiple times for multiple keys. - Python:
Copy code
from prefect.deployments import run_deployment

    run_deployment(
        name="<FLOW>/<DEPLOYMENT>",
        job_variables={"env": {"AIRBYTE_TARGET_ENV": "staging"}}
    )
CLI help: 
prefect deployment run --help
4) If you must keep env at the deployment level - You’ll need to update it by redeploying when values change: - With a project/prefect.yaml: 
prefect deploy
- Or in Python: 
flow.from_source(...).deploy(..., job_variables={...})
Which should you use? - If the env applies to all (or most) deployments in a pool and is truly “infrastructure-level” (e.g., base URLs, default resources): put it in the Work Pool. - If it’s application/toggle-like and needs frequent updates per environment/tenant: use Prefect Variables read in your flow. - For occasional exceptions: override per-run with 
--job-variable
or 
run_deployment(job_variables=...)
. - For secrets, use your secret manager or provider’s native secret injection (and reference them in your worker templates); don’t store secrets in Prefect Variables. Helpful docs - Variables: Concepts • How-to guide - Job variables: Customize job variables - Prefect YAML / Deploy: prefect.yaml how-to A couple of clarifying questions so I can recommend the cleanest setup: - Which worker type are you using (Process, Docker, Kubernetes)? - Are these env values global across all your Airbyte syncs, or flow/deployment specific? - Are any of these values secrets?
v
@Marvin How to pass namesapce and image name in the while cretaing workpool via get_client method await client.create_work_pool(
m
✅ thought for 415.9 seconds
In Prefect 3, you set things like Kubernetes namespace and image on the Work Pool’s base job template. When creating the work pool via 
get_client
, pass a 
base_job_template
dict whose 
job_configuration
includes 
namespace
and 
image
. If you’re creating a Kubernetes work pool: - Make sure the Kubernetes integration is installed (
pip install 'prefect[kubernetes]'
). - Grab the default base job template for the “kubernetes” worker type, update the fields, then create the pool. Example:
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.workers.utilities import get_default_base_job_template_for_infrastructure_type

async def main():
    # Requires `prefect[kubernetes]`
    base = get_default_base_job_template_for_infrastructure_type("kubernetes")
    if base is None:
        raise RuntimeError("Kubernetes worker type not available. Install with: pip install 'prefect[kubernetes]'")

    # Set your defaults here
    base["job_configuration"]["namespace"] = "my-namespace"
    base["job_configuration"]["image"] = "myrepo/prefect:3.1.0"

    async with get_client() as client:
        await client.create_work_pool(
            name="k8s-pool",
            type="kubernetes",
            base_job_template=base,
            description="K8s pool with custom image and namespace",
        )

asyncio.run(main())
Notes - Those values become the defaults for all deployments running on this pool. You can still override per-deployment (job variables) or per-run. - For Docker or other worker types, use the corresponding type and keys. For example, Docker pools support 
image
in 
job_configuration
as well:
Copy code
base = get_default_base_job_template_for_infrastructure_type("docker")
  base["job_configuration"]["image"] = "myrepo/app:tag"
Helpful references - Work pools: Concepts • How-to guide - Kubernetes integration: Prefect Kubernetes Quick check: Are you creating a Kubernetes pool? If so, the snippet above should work once 
prefect[kubernetes]
is installed. If you’re using a different worker type, tell me which one and I’ll tailor the 
job_configuration
keys for you.
v
@Marvin The above is not working as expected the give namespace and image name is not filled in the workpool base = await get_default_base_job_template_for_infrastructure_type(type) if base: base["job_configuration"]["namespace"] = namespace base["job_configuration"]["image"] = image payload.base_job_template = base Check and comback again
m
✅ thought for 48.1 seconds
Thanks for the snippet — a few gotchas can cause exactly what you’re seeing. Likely causes - Worker type mismatch: If you created a “process” work pool, fields like namespace/image don’t apply. Use type="kubernetes" (or "docker"). - Wrong place to set values: The UI’s “Default job variables” form is driven by the template’s variables schema defaults, not the job_configuration values. Setting only job_configuration won’t show in the UI unless the job_configuration references those variables. - Template missing references: If job_configuration doesn’t reference the variables (e.g., "{{ namespace }}"), changes to defaults won’t flow through. - Minor: 
get_default_base_job_template_for_infrastructure_type
is synchronous; don’t await it. What to do 1) Make sure you’re creating a Kubernetes work pool and have the integration installed: - pip install 'prefect[kubernetes]' 2) Set defaults on the variables schema and ensure job_configuration references them: - Set defaults under base["variables"]["properties"][...]["default"] - Ensure base["job_configuration"]["..."] = "{{ ... }}" so the variables are used Corrected example
Copy code
import asyncio
from prefect.client.orchestration import get_client
from prefect.workers.utilities import get_default_base_job_template_for_infrastructure_type
from prefect.client.schemas.actions import WorkPoolCreate

async def create_k8s_pool(name: str, namespace: str, image: str):
    # This function is synchronous
    base = get_default_base_job_template_for_infrastructure_type("kubernetes")
    if not base:
        raise RuntimeError("Kubernetes worker type not available. Install with: pip install 'prefect[kubernetes]'")

    # 1) Set defaults in the variables schema
    props = base.setdefault("variables", {}).setdefault("properties", {})
    props.setdefault("namespace", {"type": "string"})
    props.setdefault("image", {"type": "string"})
    props["namespace"]["default"] = namespace
    props["image"]["default"] = image

    # 2) Ensure job_configuration uses those variables
    base.setdefault("job_configuration", {})
    base["job_configuration"]["namespace"] = "{{ namespace }}"
    base["job_configuration"]["image"] = "{{ image }}"

    async with get_client() as client:
        # Either pass a WorkPoolCreate...
        await client.create_work_pool(
            work_pool=WorkPoolCreate(
                name=name,
                type="kubernetes",
                base_job_template=base,
                description="K8s pool with custom image and namespace",
            )
        )
        # ...or (depending on your SDK version) as kwargs:
        # await client.create_work_pool(
        #     name=name,
        #     type="kubernetes",
        #     base_job_template=base,
        #     description="K8s pool with custom image and namespace",
        # )

asyncio.run(create_k8s_pool("k8s-pool", "my-namespace", "myrepo/prefect:3.1.0"))
CLI alternative (useful to inspect the exact schema) - Dump the default template, edit defaults, then update the pool:
Copy code
prefect work-pool get-default-base-job-template --type kubernetes > base.json
# edit base.json:
# - set variables.properties.namespace.default
# - set variables.properties.image.default
# - ensure job_configuration.namespace = "{{ namespace }}"
# - ensure job_configuration.image = "{{ image }}"

prefect work-pool create "k8s-pool" --type kubernetes --base-job-template base.json
# or update an existing pool:
prefect work-pool update "k8s-pool" --base-job-template base.json
Double-checks - Confirm the pool type you’re creating (kubernetes vs docker vs process). - In the UI, open Work Pools > your pool > Default job variables — you should see namespace and image populated from the variables defaults. - If you share your 
payload
(WorkPoolCreate) dict and worker type, I can spot-check the exact keys for you. Docs - Work pools how-to: Manage work pools
- Kubernetes integration: Prefect Kubernetes - CLI: Work pool CLI If you’re on Docker instead of K8s, swap type="docker" and set image in the same pattern (Docker pools don’t have namespace).
v
@Marvin How to set it up with the entrypoint.sh file along with worker start command to do it
m
✅ thought for 99.2 seconds
Got it — here’s a practical way to set up the work pool (namespace, image) and then start the worker from an entrypoint.sh. This approach patches the base job template at container start, so you don’t need to pre-bake it. Assumptions - You’re targeting a Kubernetes work pool (namespace + image); for Docker, see notes below. - Your image has the integration installed: 
pip install 'prefect[kubernetes]'
(baked into the image). - Prefect API key/URL are provided via env vars (
PREFECT_API_URL
, 
PREFECT_API_KEY
). entrypoint.sh
Copy code
#!/usr/bin/env bash
set -euo pipefail

# Required env
: "${PREFECT_API_URL:?Set PREFECT_API_URL}"
: "${PREFECT_API_KEY:?Set PREFECT_API_KEY}"

# Customizable env
: "${POOL_NAME:=k8s-pool}"
: "${POOL_TYPE:=kubernetes}"          # use "docker" for Docker workers
: "${WORKER_NAME:=$(hostname)}"

# K8s-specific defaults
: "${K8S_NAMESPACE:=default}"
: "${WORKER_IMAGE:=prefecthq/prefect:3-latest}"

echo "[entrypoint] Preparing base job template for pool type: ${POOL_TYPE}"

# Build a patched base job template using Python to avoid jq dependency
PATCHED_JSON="$(python - <<'PY'
import json, os, sys
from prefect.workers.utilities import get_default_base_job_template_for_infrastructure_type

pool_type = os.environ.get("POOL_TYPE", "kubernetes")
base = get_default_base_job_template_for_infrastructure_type(pool_type)
if not base:
    print("{}", end="")
    sys.exit(1)

props = base.setdefault("variables", {}).setdefault("properties", {})
job_cfg = base.setdefault("job_configuration", {})

# For Kubernetes: set namespace + image defaults and wire them into job_configuration
if pool_type == "kubernetes":
    ns = os.environ.get("K8S_NAMESPACE", "default")
    img = os.environ.get("WORKER_IMAGE", "prefecthq/prefect:3-latest")
    props.setdefault("namespace", {"type": "string"})
    props.setdefault("image", {"type": "string"})
    props["namespace"]["default"] = ns
    props["image"]["default"] = img
    job_cfg["namespace"] = "{{ namespace }}"
    job_cfg["image"] = "{{ image }}"

# For Docker: set image default
elif pool_type == "docker":
    img = os.environ.get("WORKER_IMAGE", "prefecthq/prefect:3-latest")
    props.setdefault("image", {"type": "string"})
    props["image"]["default"] = img
    job_cfg["image"] = "{{ image }}"

json.dump(base, sys.stdout)
PY
)"

if [ -z "${PATCHED_JSON}" ] || [ "${PATCHED_JSON}" = "{}" ]; then
  echo "[entrypoint] Failed to build base job template. Ensure the integration for ${POOL_TYPE} is installed."
  exit 1
fi
echo "${PATCHED_JSON}" > /tmp/base-job-template.json

echo "[entrypoint] Creating/updating work pool: ${POOL_NAME}"
# --overwrite lets us update an existing pool in-place
prefect work-pool create "${POOL_NAME}" \
  --type "${POOL_TYPE}" \
  --base-job-template /tmp/base-job-template.json \
  --overwrite

echo "[entrypoint] Starting worker for pool ${POOL_NAME} as ${WORKER_NAME}"
exec prefect worker start --pool "${POOL_NAME}" --name "${WORKER_NAME}"
Dockerfile snippet
Copy code
FROM prefecthq/prefect:3-latest
# For Kubernetes workers:
RUN pip install 'prefect[kubernetes]'
COPY entrypoint.sh /entrypoint.sh
RUN chmod +x /entrypoint.sh
ENTRYPOINT ["/entrypoint.sh"]
Environment you’ll typically set when running this container - Required: - 
PREFECT_API_URL
, 
PREFECT_API_KEY
- Optional/custom: - 
POOL_NAME=k8s-pool
- 
POOL_TYPE=kubernetes
(or 
docker
) - 
K8S_NAMESPACE=your-namespace
(K8s only) - 
WORKER_IMAGE=yourrepo/yourimage:tag
- 
WORKER_NAME=my-worker-1
Why this works - It fetches the correct default base job template for the worker type, then: - Sets defaults in the variables schema (so values are visible/editable in the UI’s “Default job variables”). - Ensures the job configuration references those variables via 
{{ ... }}
placeholders. - It creates/overwrites the work pool with the patched template and finally starts the worker. Adapting for Docker - Set 
POOL_TYPE=docker
- You only need `WORKER_IMAGE`; Docker doesn’t use 
namespace
. - The script already handles these conditionally. Helpful docs - Work pools CLI: Work pool CLI - Worker start CLI: Worker CLI - Kubernetes integration: Prefect Kubernetes - Managing work pools: Manage work pools If you share which worker type you’re targeting and the exact env you want to inject, I can tailor the template patching (e.g., imagePullPolicy, resources, service account, etc.).
v
@Marvin I'm creating a worker with .sh file in the with below config I want to create an workpool with namespace and image name before starting worker in .sh file can we able to do it ? prefect worker start --pool "$WORK_POOL_NAME" --type kubernetes prefect work-pool set-defaults "$WORK_POOL_NAME" \ --image alert_validator --namespace redflare-v2-core-api
m
✅ thought for 42.6 seconds
Yes — do it in your entrypoint.sh before starting the worker. There isn’t a 
prefect work-pool set-defaults
command; instead, you create/update the pool with a base job template that sets namespace and image. Then start the worker for that pool. Also, you don’t need 
--type
on `prefect worker start`; the pool’s type defines the worker type. Here’s a self-contained entrypoint that: - Fetches the default Kubernetes base job template - Patches it with your namespace and image defaults (and wires them into job_configuration) - Creates/updates the work pool - Starts the worker
Copy code
#!/usr/bin/env sh
set -euo pipefail

: "${PREFECT_API_URL:?Set PREFECT_API_URL}"
: "${PREFECT_API_KEY:?Set PREFECT_API_KEY}"

WORK_POOL_NAME="${WORK_POOL_NAME:-k8s-pool}"
K8S_NAMESPACE="${K8S_NAMESPACE:-redflare-v2-core-api}"
WORKER_IMAGE="${WORKER_IMAGE:-alert_validator}"

# 1) Get default K8s base job template (requires 'prefect[kubernetes]' in the image)
prefect work-pool get-default-base-job-template --type kubernetes > /tmp/base.json

# 2) Patch template to set defaults and wire placeholders
python - <<'PY'
import json, os
path = "/tmp/base.json"
with open(path) as f:
    base = json.load(f)

props = base.setdefault("variables", {}).setdefault("properties", {})
jc = base.setdefault("job_configuration", {})

# Ensure variables exist and set defaults
props.setdefault("namespace", {"type": "string"})
props.setdefault("image", {"type": "string"})
props["namespace"]["default"] = os.environ["K8S_NAMESPACE"]
props["image"]["default"] = os.environ["WORKER_IMAGE"]

# Ensure job_configuration uses the variables
jc["namespace"] = "{{ namespace }}"
jc["image"] = "{{ image }}"

with open(path, "w") as f:
    json.dump(base, f)
PY

# 3) Create/update the work pool with this template
prefect work-pool create "${WORK_POOL_NAME}" \
  --type kubernetes \
  --base-job-template /tmp/base.json \
  --overwrite

# 4) Start the worker for that pool (no --type needed)
exec prefect worker start --pool "${WORK_POOL_NAME}"
Notes - Make sure your container includes the Kubernetes extra: 
pip install 'prefect[kubernetes]'
. - Use a fully qualified image name for WORKER_IMAGE if needed (e.g., 
us-east1-docker.pkg.dev/proj/repo/alert_validator:tag
). - If you only need to update an existing pool, you can swap the create step with:
Copy code
prefect work-pool update "${WORK_POOL_NAME}" --base-job-template /tmp/base.json
CLI references: - Work pool CLI: Work pool CLI - Worker CLI: Worker CLI If you prefer a Docker worker instead of K8s, set 
--type docker
in the create step and only set 
image
in the template; Docker doesn’t use 
namespace
.
v
@Marvin On prefect server startup how do I create workpool so the worker can start their worker with workpool • I have to creat 10 workpool currently we are having work pool name in the .env which is not ideal if the number of workpool increases • How can I create workpool with while starting prefect server is this right approach to do it ??
m
✅ thought for 246.3 seconds
Great question. Short answer: there isn’t a built-in “on server startup” hook that auto-creates work pools. The recommended pattern is to run an idempotent bootstrap step right after the API is up that “ensures” your work pools exist (creates them if missing, updates if present). This scales well beyond a few pools and avoids hardcoding names in a .env. Two common approaches you can choose from: 1) CLI approach (simple, idempotent) - Use 
prefect work-pool create NAME --type TYPE --overwrite
in a startup script. The 
--overwrite
flag makes it idempotent. - Example bootstrap script that reads from a config file: pools.yaml
Copy code
work_pools:
  - name: etl
    type: process
    description: "ETL pool"
    paused: false
  - name: ml-jobs
    type: process
    paused: true
bootstrap_pools.sh
Copy code
#!/usr/bin/env bash
set -euo pipefail

# Wait for the API to be ready
until curl -sf "${PREFECT_API_URL%/}/health" >/dev/null; do
  echo "Waiting for Prefect API at $PREFECT_API_URL..."
  sleep 2
done

# Create/update work pools
# Requires yq (<https://mikefarah.gitbook.io/yq/>)
len=$(yq '.work_pools | length' pools.yaml)
for i in $(seq 0 $((len - 1))); do
  name=$(yq -r ".work_pools[$i].name" pools.yaml)
  type=$(yq -r ".work_pools[$i].type" pools.yaml)
  desc=$(yq -r ".work_pools[$i].description // \"\"" pools.yaml)
  paused=$(yq -r ".work_pools[$i].paused // false" pools.yaml)

  # Create or update if exists
  if [ -n "$desc" ] && [ "$desc" != "null" ]; then
    prefect work-pool create "$name" --type "$type" --description "$desc" ${paused:+--paused} --overwrite
  else
    prefect work-pool create "$name" --type "$type" ${paused:+--paused} --overwrite
  fi
done
- If you’re using Docker/Kubernetes pools and need to provision infrastructure, you can follow up with:
Copy code
prefect work-pool provision-infrastructure "$name"
Note: install extras when needed (e.g., 
uv run --with 'prefect[kubernetes]' prefect work-pool provision-infrastructure --help
). 2) Python API approach (flexible, good for custom logic) - Use the client to “get or create” from a config. This runs anywhere you can point 
PREFECT_API_URL
at your server. pools.yaml
Copy code
work_pools:
  - name: etl
    type: process
    description: "ETL pool"
    concurrency_limit: 10
  - name: ml-jobs
    type: process
    concurrency_limit: 5
ensure_work_pools.py
Copy code
import asyncio
import yaml
from pathlib import Path

from prefect import get_client
from prefect.client.schemas.actions import WorkPoolCreate, WorkPoolUpdate
from prefect.exceptions import ObjectNotFound

CONFIG_PATH = Path("pools.yaml")

async def ensure_work_pools():
    config = yaml.safe_load(CONFIG_PATH.read_text())
    pools = config.get("work_pools", [])

    async with get_client() as client:
        for p in pools:
            name = p["name"]
            wp_type = p["type"]
            description = p.get("description")
            concurrency = p.get("concurrency_limit")
            paused = p.get("paused")

            try:
                # Check if the pool exists
                await client.read_work_pool(name)
                # Update if desired fields are provided
                update = WorkPoolUpdate(
                    description=description,
                    concurrency_limit=concurrency,
                    is_paused=paused,
                )
                await client.update_work_pool(name, update)
                print(f"Updated work pool: {name}")
            except ObjectNotFound:
                # Create if missing
                create = WorkPoolCreate(
                    name=name,
                    type=wp_type,
                    description=description,
                    concurrency_limit=concurrency,
                    is_paused=paused,
                )
                await client.create_work_pool(create)
                print(f"Created work pool: {name}")

if __name__ == "__main__":
    asyncio.run(ensure_work_pools())
Then wire this into your startup: - Docker Compose: run the script in an init container or as part of the server’s entrypoint after the API is healthy. - Kubernetes: add a postStart hook or a Job that runs after the API is reachable. Starting workers without hardcoding pool names - You can still source a config and start workers per pool dynamically: start_workers.sh
Copy code
#!/usr/bin/env bash
set -euo pipefail

len=$(yq '.work_pools | length' pools.yaml)
for i in $(seq 0 $((len - 1))); do
  name=$(yq -r ".work_pools[$i].name" pools.yaml)
  prefect worker start --pool "$name" --name "worker-$name" &
done
wait
Notes and best practices - There’s no server-side hook to auto-create pools at process start; using an idempotent bootstrap is the right approach. - Use a single source of truth (YAML/JSON) for your work pools and have both the “ensure pools” step and “start workers” step read from it. - For non-process pools (docker, kubernetes, ecs), you may also want to manage base job templates. The CLI supports 
--base-job-template
on create, and the Python client supports setting 
base_job_template
within 
WorkPoolCreate
. - If you’re on Prefect 2.x, consider upgrading to 3.x; commands above are for Prefect 3.x. Helpful docs - Work pools concepts: Work pools (Prefect 3) - CLI reference: prefect work-pool CLI - REST API for work pools: Work pools REST API - Worker CLI: prefect worker start A couple of questions so I can tailor an example to your setup: - Where are you running Prefect Server (Docker Compose, Kubernetes, bare VM)? - Which work pool types do you need (process, docker, kubernetes, etc.) and do you use custom base job templates? - Do you want a single script to both “ensure” pools and start N workers per pool?
v
@Marvin prefect check and create block if block bucket doesn' exist currently creating like this I want check if bucket name exist if exist do nothing otheriwse create it await S3Bucket( bucket_name="redflare-v2-dev-bucket", bucket_folder="flow_results", ).save("flow-result-dev")
m
✅ thought for 258.7 seconds
Do you want to check for the Prefect block’s existence or the actual S3 bucket in AWS? I’ll assume you mean the Prefect block. Here’s an idempotent “get-or-create” pattern in Prefect 3.x for an S3Bucket block. Sync version
Copy code
from prefect_aws.s3 import S3Bucket

BLOCK_NAME = "flow-result-dev"

try:
    # If the block exists, this succeeds and you do nothing
    S3Bucket.load(BLOCK_NAME)
    print(f"Block '{BLOCK_NAME}' already exists; doing nothing.")
except ValueError:
    # If it doesn't exist, create and save it
    s3_block = S3Bucket(
        bucket_name="redflare-v2-dev-bucket",
        bucket_folder="flow_results",
        # credentials=AwsCredentials.load("my-aws-creds")  # if you use a creds block
    )
    s3_block.save(BLOCK_NAME)
    print(f"Created block '{BLOCK_NAME}'.")
Async version (if your codebase is async)
Copy code
from prefect_aws.s3 import S3Bucket

BLOCK_NAME = "flow-result-dev"

try:
    S3Bucket.load(BLOCK_NAME)
    print(f"Block '{BLOCK_NAME}' already exists; doing nothing.")
except ValueError:
    await S3Bucket(
        bucket_name="redflare-v2-dev-bucket",
        bucket_folder="flow_results",
    ).save(BLOCK_NAME)
    print(f"Created block '{BLOCK_NAME}'.")
Notes - 
S3Bucket.load(name)
raises a ValueError if the block doesn’t exist, so catching that is the simplest “exists or create” check. - If you ever want to force-update the block to match your latest config, pass 
overwrite=True
to 
save(...)
. Otherwise, leave it out to “do nothing if it exists.” - If you meant checking the actual AWS bucket’s existence instead, use boto3 (
head_bucket
and 
create_bucket
) before saving the block; I can share a snippet for that too. Docs - Blocks overview: Prefect blocks - AWS integration: prefect-aws integration
v
@Marvin don't we have any filter kind like we do for flow like read_flows with flow_filter
m
✅ thought for 264.0 seconds
Short answer: there isn’t a high-level “read_blocks(filter=…)” convenience like there is for flows. For blocks you generally either: - Load by name using the block class (simple, idempotent for “exists or create” use-cases) - Read a specific block by name + type via the client - Use the lower-level block document filters via the REST API if you truly need list/search behavior Practical options 1) Recommended for “exists or create”
Copy code
from prefect_aws.s3 import S3Bucket

name = "flow-result-dev"

try:
    S3Bucket.load(name)  # raises ValueError if not found
    print("Block exists")
except ValueError:
    S3Bucket(bucket_name="redflare-v2-dev-bucket", bucket_folder="flow_results").save(name)
    print("Block created")
2) Read a specific block by name and type via the client
Copy code
import asyncio
from prefect import get_client
from prefect.exceptions import ObjectNotFound

async def main():
    async with get_client() as client:
        try:
            # block_type_slug for S3Bucket is typically "s3-bucket"
            await client.read_block_document_by_name(name="flow-result-dev", block_type_slug="s3-bucket")
            print("Block exists")
        except ObjectNotFound:
            print("Block not found")

asyncio.run(main())
3) Filtering (advanced) - Prefect exposes filter models for blocks (e.g., BlockDocumentFilterName, BlockTypeFilterSlug) in 
prefect.client.schemas.filters
, similar to flow filters. - However, there isn’t a single SDK call like 
read_blocks(filter=...)
. To list/filter blocks, you’d call the block-documents filter endpoint and pass those filter objects. If you want this, I can share a concrete example using the client to call the REST “block documents” filter endpoint. Docs - Blocks overview: Blocks (Prefect 3) - Filters reference: Filter schemas - Client reference: PrefectClient
v
@Marvin create prefect worker pool with namespace and image name with base job template in prefect cli via entrypoint.sh
m
✅ thought for 79.7 seconds
Yes—create a Kubernetes work pool with a base job template that sets your namespace and image, all from an entrypoint.sh. Here’s a practical, idempotent example you can drop into your container: Requirements - curl and jq installed in your image - Prefect 3.x - If you’ll run a Kubernetes worker from this container, ensure your image has prefect[kubernetes] installed entrypoint.sh
Copy code
#!/usr/bin/env bash
set -euo pipefail

# Config via env vars
: "${PREFECT_API_URL:=<http://prefect-server:4200/api>}"
: "${POOL_NAME:=k8s-pool}"
: "${K8S_NAMESPACE:=default}"
: "${JOB_IMAGE:=myrepo/myimage:tag}"
: "${SERVICE_ACCOUNT:=prefect-worker}"
: "${IMAGE_PULL_POLICY:=IfNotPresent}"

# Wait for Prefect API to be ready
until curl -sf "${PREFECT_API_URL%/}/health" >/dev/null; do
  echo "Waiting for Prefect API at $PREFECT_API_URL..."
  sleep 2
done

# Get the default Kubernetes base job template and set your defaults
tmp="$(mktemp)"
prefect work-pool get-default-base-job-template --type kubernetes > "$tmp"

jq \
  --arg ns "$K8S_NAMESPACE" \
  --arg img "$JOB_IMAGE" \
  --arg sa "$SERVICE_ACCOUNT" \
  --arg ipp "$IMAGE_PULL_POLICY" \
  '
  .variables.properties.namespace.default = $ns
  | .variables.properties.image.default = $img
  | .variables.properties.service_account_name.default = $sa
  | .variables.properties.image_pull_policy.default = $ipp
  ' "$tmp" > /tmp/base-job-template.json

# Create or update the work pool with the customized base job template
prefect work-pool create "$POOL_NAME" \
  --type kubernetes \
  --base-job-template /tmp/base-job-template.json \
  --overwrite

# Optional: start a worker for this pool
# exec prefect worker start --pool "$POOL_NAME" --name "${WORKER_NAME:-worker-$POOL_NAME}"
Notes - Using defaults in the base job template lets you keep these values override-able per deployment if needed. If you want to hardcode them into the manifest instead, you can replace the placeholders in job_configuration directly with jq. - For Docker pools, swap the type and template: 
prefect work-pool get-default-base-job-template --type docker
and set 
.variables.properties.image.default = "..."
. - You can add 
--paused
or set concurrency on the pool with 
prefect work-pool set-concurrency-limit
. Docs - Work pools (concepts): Work pools - CLI reference: prefect work-pool CLI - Worker CLI: prefect worker start If you share your runtime environment (Docker Compose, Kubernetes, etc.), I can tailor the script further (e.g., init container vs. postStart hook, multiple pools, etc.).
v
@Marvin I'm planning to create workflow module on top prefect orchestration. With an example of nodes and edges a node can have a branched for instance If I consider each node as either task or flow depends of the complexity of each node. If I have 5 branches for a node how can I execute all of it's edges like task in parallel. simply how would I execute multiple tasks or flows at once ?
m
✅ thought for 857.0 seconds
Great question — you can absolutely model a node that fans out to 5 branches and run all branches concurrently in Prefect 3.x. Here are the main patterns you can use depending on whether each branch is a task (same process) or a flow (separate run). Assumptions: Prefect 3.x Run multiple tasks in parallel (same process) - Use 
task.submit(...)
to fan out and Prefect futures to fan in. - Configure a task runner on the flow to control concurrency. Threads are good for I/O-bound work; processes are better for CPU-bound work. Example: 5 branches in parallel with a fan-in
Copy code
from prefect import flow, task
from prefect.task_runners import ThreadPoolTaskRunner  # or ProcessPoolTaskRunner
from prefect.futures import wait

@task
def do_branch(i: int) -> str:
    # do work for branch i
    return f"done {i}"

@flow(task_runner=ThreadPoolTaskRunner(max_workers=10))
def parent():
    # fan-out
    futures = [do_branch.submit(i) for i in range(5)]
    # fan-in: wait or gather results
    wait(futures)  # optional if you just call .result() below
    results = [f.result() for f in futures]
    return results

if __name__ == "__main__":
    print(parent())
- For CPU-heavy branches, switch to processes:
Copy code
@flow(task_runner=ProcessPoolTaskRunner(max_workers=5))
def parent():
    ...
- You can also use mapping instead of a loop, especially when passing shared args:
Copy code
from prefect import flow, task, unmapped

@task
def process_item(item, cfg): ...

@flow
def parent(items, cfg):
    futures = process_item.map(items, cfg=unmapped(cfg))
    return futures.result()  # blocks and returns list of results
Run multiple flows in parallel (separate infrastructure) Subflows called directly in code run in the same process and do not run concurrently. If you want true parallelism across flows, trigger multiple flow runs of a deployment. Pattern A: Trigger deployed flows concurrently from a task (simple, synchronous)
Copy code
from prefect import flow, task
from prefect.deployments import run_deployment

@task
def trigger_branch(dep_name: str, params: dict | None = None):
    # Runs on separate infra; returns when the child flow run finishes
    return run_deployment(dep_name, parameters=params, as_subflow=True)

@flow
def orchestrator():
    deps = [("project/branch_flow:prod", {"i": i}) for i in range(5)]
    futures = [trigger_branch.submit(dep, params) for dep, params in deps]
    return [f.result() for f in futures]
Pattern B: Use the async client API for non-blocking fan-out/fan-in
Copy code
import asyncio
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.flow_runs import wait_for_flow_run

@flow
async def orchestrator_async(deployment_ids: list[str]):
    async with get_client() as client:
        created = []
        for i, dep_id in enumerate(deployment_ids):
            run = await client.create_flow_run_from_deployment(
                deployment_id=dep_id,
                parameters={"i": i},
            )
            created.append(run)
        # fan-in: wait on all runs
        await asyncio.gather(*(wait_for_flow_run(r.id) for r in created))
Notes and tips - Threads vs processes: ThreadPoolTaskRunner is best for I/O-bound work. For CPU-bound branches, prefer ProcessPoolTaskRunner or use a distributed runner (Dask/Ray). - Dependencies: Pass futures as inputs to downstream tasks; Prefect will wait automatically (no manual 
result()
needed). - Subflows: Calling 
subflow()
within a flow is sequential in the same process. To truly run flows in parallel, trigger separate runs (deployments) via 
run_deployment
or the async client pattern above. - Distributed options: If you need to scale beyond a single machine, consider Dask or Ray task runners. Docs you may find helpful - Run work concurrently - Task runners
- Prefect futures API - Run deployments programmatically - Dask integration A couple of clarifying questions so I can tailor the pattern: - Will each branch be a task or its own flow? - Are branches I/O-bound or CPU-bound? - Do you want all branches to run on the same machine/process, or on separate workers/infrastructure?
v
@Marvin can I have deployment like this - grammar-detection-deployment will it work if deployment read by name ?
m
✅ thought for 370.8 seconds
Short answer: yes, you can name a deployment 
grammar-detection-deployment
(hyphens are fine). But when you “read by name,” you need the fully qualified name: 
flow_name/deployment_name
. The deployment name alone is not globally unique. Examples - Python:
Copy code
from prefect.deployments import run_deployment

# Use the fully qualified name
run_deployment("my-flow/grammar-detection-deployment", parameters={"x": 1})

# Or use the deployment ID (UUID)
run_deployment("d3f1b3ef-1234-5678-9abc-0123456789de")
- CLI:
Copy code
prefect deployment run "my-flow/grammar-detection-deployment" --watch
Helpful docs: - Run deployments programmatically - Deployments concept If you share your flow name, I can show the exact string to use.
v
@Marvin I get error object not found when reading by name like this prefect.exceptions.PrefectHTTPStatusError: Client error '404 Not Found' for url 'https://flows.services.stratforge.dev/api/deployments/name/grammar-detection/grammar_detection_deployment' Response: {'detail': 'Deployment not found'}
m
✅ thought for 52.1 seconds
404 “Deployment not found” means the pair you’re passing does not match any deployment in your current workspace. A few common gotchas: - The name must be fully qualified as 
flow_name/deployment_name
and must match exactly (hyphen vs underscore often trips people up). - Your flow’s registered name may not be what you expect if you set 
@flow(name="...")
or 
flow.with_options(name="...")
. - You might be on a different profile/workspace than where the deployment lives. Quick checks 1) List what Prefect thinks the names are:
Copy code
prefect deployment ls
# or filter to a flow if you think you know it
prefect deployment ls --flow-name "grammar_detection"
prefect deployment ls --flow-name "grammar-detection"
2) Try the exact string from the listing in `run_deployment`:
Copy code
from prefect.deployments import run_deployment
run_deployment("ACTUAL_FLOW_NAME/ACTUAL_DEPLOYMENT_NAME")
3) If you prefer to bypass naming issues, use the deployment ID (UUID) shown in `prefect deployment ls`:
Copy code
run_deployment("d3f1b3ef-1234-5678-9abc-0123456789de")
4) Confirm you’re in the right workspace/profile:
Copy code
prefect profile inspect
Why your example likely failed Your URL shows 
.../deployments/name/grammar-detection/grammar_detection_deployment
. If your flow is actually registered as 
grammar_detection
(underscore), the correct name would be 
grammar_detection/grammar_detection_deployment
, not 
grammar-detection/...
. The hyphen/underscore mismatch will produce a 404. Docs for reference: - Run deployments programmatically - Deployments concept If you paste the output of 
prefect deployment ls
for that flow, I can give you the exact string to use.
v
@Marvin while creating workpool how can I inject namespace and image name and variable to base job template I'm using custom K8_base_job_Teamplate
m
✅ thought for 347.3 seconds
Got it! Assuming you’re on Prefect 3.x and using a custom Kubernetes base job template, you inject things like namespace, image, and env via Jinja-style placeholders in the template and a variables schema. Then you set defaults at the work pool level (via the template) and override per deployment with 
job_variables
. Here’s the minimal pattern to follow. 1) Author a base job template (JSON) - Define placeholders in 
job_configuration
and 
job_manifest
- Declare them in the 
variables
schema (with optional defaults) Example:
Copy code
{
  "job_configuration": {
    "namespace": "{{ namespace }}",
    "job_manifest": {
      "apiVersion": "batch/v1",
      "kind": "Job",
      "metadata": {
        "generateName": "{{ name }}-",
        "labels": "{{ labels }}",
        "namespace": "{{ namespace }}"
      },
      "spec": {
        "backoffLimit": "{{ backoff_limit }}",
        "template": {
          "spec": {
            "restartPolicy": "Never",
            "serviceAccountName": "{{ service_account_name }}",
            "containers": [
              {
                "name": "prefect-job",
                "image": "{{ image }}",
                "imagePullPolicy": "{{ image_pull_policy }}",
                "args": "{{ command }}",
                "env": "{{ env }}"
              }
            ]
          }
        }
      }
    }
  },
  "variables": {
    "type": "object",
    "properties": {
      "namespace": {
        "type": "string",
        "default": "default",
        "description": "Kubernetes namespace"
      },
      "image": {
        "type": "string",
        "description": "Container image (e.g. <http://ghcr.io/org/app:tag|ghcr.io/org/app:tag>)"
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "service_account_name": {
        "type": "string",
        "default": null
      },
      "backoff_limit": {
        "type": "integer",
        "default": 0
      },
      "env": {
        "type": [
          "object",
          "array",
          "null"
        ],
        "default": null,
        "description": "Dict of env vars or full Kubernetes env list"
      }
    }
  }
}
Notes: - Use 
{{ namespace }}
, 
{{ image }}
, and 
{{ env }}
where you want those values to land. - 
env
can be: - a dict: 
{"LOG_LEVEL": "DEBUG", "ENV": "staging"}
- or a full Kubernetes list: 
[{"name":"LOG_LEVEL","value":"DEBUG"}, {"name":"SECRET", "valueFrom": {...}}]
- Prefect will inject 
name
, 
labels
, 
command
, and core Prefect env automatically; your values are merged in. 2) Create or update the work pool with your template - Pool type is 
kubernetes
Create:
Copy code
prefect work-pool create "my-k8s-pool" --type kubernetes --base-job-template ./base-job-template.json
Update:
Copy code
prefect work-pool update "my-k8s-pool" --base-job-template ./base-job-template.json
Inspect:
Copy code
prefect work-pool inspect "my-k8s-pool" --output json
3) Override per deployment with job_variables In code, pass 
job_variables
when creating the deployment. Example:
Copy code
from prefect import flow, serve

@flow
def my_flow():
    print("hi from k8s")

if __name__ == "__main__":
    serve(
        my_flow.to_deployment(
            name="my-deployment",
            work_pool_name="my-k8s-pool",
            job_variables={
                "namespace": "staging",
                "image": "<http://ghcr.io/my-org/my-image:1.2.3|ghcr.io/my-org/my-image:1.2.3>",
                "image_pull_policy": "Always",
                "service_account_name": "my-sa",
                "env": {
                    "ENV": "staging",
                    "LOG_LEVEL": "DEBUG"
                }
            }
        )
    )
That’s it: defaults live in the base job template’s 
variables.properties[*].default
, and you override per deployment via 
job_variables
. The worker fills in Prefect-managed bits at runtime.
Would you like me to review your current K8_base_job_Teamplate and suggest exact placeholders and schema updates for your use case? If you can paste it here, I’ll annotate it. Helpful docs: - Work pools: Work pools - Kubernetes worker guide: Kubernetes worker
v
@Marvin Currently We are creating and storing ENV variable inside workpool via manual approach ( which could be prone to error or something) What approach should we follow ? What are best practices to store and use environment variables and secrets for the workpool the job can use it ?
m
✅ thought for 267.8 seconds
Great question. Short answer: don’t hand-enter env vars into the work pool. Put references to config and secrets in the pool’s base job template, and source their values from Kubernetes Secrets, ConfigMaps, and/or Prefect Secret blocks. Use deployment 
job_variables
only for non-sensitive, per-deployment overrides. Recommended approach (Prefect 3.x + Kubernetes): - Secrets - Prefer Kubernetes Secrets via 
envFrom
or 
valueFrom
in the base job template; the secret values never touch Prefect. - Or use Prefect Secret blocks in the template with `{{ prefect.blocks.secret.<name> }}`; values are pulled at runtime by the worker and injected into the pod env. - Do not store secrets in deployment 
job_variables
or in Prefect Variables. - Non-sensitive config - Use Prefect Variables (
{{ prefect.variables.<name> }}
) or Kubernetes ConfigMaps (
valueFrom.configMapKeyRef
) in the base template. - Where to put things - Base job template: your secure references (K8s secret names, Prefect block/variable placeholders) and structure. - Deployment `job_variables`: only non-sensitive overrides like 
LOG_LEVEL
, 
BATCH_SIZE
, image, namespace, etc. - Merging rules - Deployment 
job_variables.env
overrides template env; Prefect also injects required env (e.g., 
PREFECT_API_URL
, flow-run ID). - Multi-environment - Use separate pools per environment (staging/prod) or make the secret name a variable in the template so deployments can pick the right one without exposing values. - Rotation - Update the K8s Secret or Prefect Secret block; no redeploy required. New runs pick up the new value. - Access & safety - Ensure the ServiceAccount used by the worker pod has RBAC to read required secrets. - Keep the template JSON in version control; update pools via CLI or UI to avoid manual edits. Starter base job template (showing best practices)
Copy code
{
  "job_configuration": {
    "namespace": "{{ namespace }}",
    "env": {
      "API_ENDPOINT": "{{ prefect.variables.api_endpoint }}",          // non-sensitive
      "DATABASE_URL": "{{ prefect.blocks.secret.db-connection }}"     // sensitive
    }
  },
  "job_manifest": {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
      "labels": "{{ labels }}",
      "namespace": "{{ namespace }}",
      "generateName": "{{ name }}-"
    },
    "spec": {
      "backoffLimit": "{{ backoff_limit }}",
      "template": {
        "spec": {
          "restartPolicy": "Never",
          "serviceAccountName": "{{ service_account_name }}",
          "containers": [
            {
              "name": "prefect-job",
              "image": "{{ image }}",
              "imagePullPolicy": "{{ image_pull_policy }}",
              "args": "{{ command }}",
              "env": "{{ env }}",
              "envFrom": [
                { "secretRef": { "name": "{{ base_secret_name }}" } }   // pulls many keys at once
              ]
            }
          ]
        }
      }
    }
  },
  "variables": {
    "type": "object",
    "properties": {
      "namespace": { "type": "string", "default": "prefect" },
      "image": { "type": "string", "default": "prefecthq/prefect:3-latest" },
      "image_pull_policy": { "type": "string", "enum": ["IfNotPresent", "Always", "Never"], "default": "IfNotPresent" },
      "service_account_name": { "type": "string", "default": "prefect-worker" },
      "backoff_limit": { "type": "integer", "default": 0 },
      "base_secret_name": { "type": "string", "default": "app-secrets" },
      "env": { "type": ["object", "array", "null"], "default": null }
    }
  }
}
Per-deployment overrides (non-sensitive only) ``` from prefect import flow, serve @flow def my_flow(): ... if name == "__main__": serve( my_flow.to_deployment( name="prod", work_pool_name="k8s-prod", job_variables={ "namespace": "prod", "image": "ghcr.io/acme/app:1.2.3", "env": {
"LOG_LEVEL": "INFO", "BATCH_SIZE": "1000" } } ) )
Copy code
Kubernetes Secrets/ConfigMaps examples
- Bulk import with envFrom (recommended when you already manage secrets in K8s):
"envFrom": [ { "secretRef": { "name": "app-secrets" } }, { "configMapRef": { "name": "app-config" } } ]
Copy code
- Per-key reference with valueFrom (fine-grained control):
"env": [ { "name": "DB_PASSWORD", "valueFrom": { "secretKeyRef": { "name": "app-secrets", "key": "db-password" } } }, { "name": "LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "app-config", "key": "log_level" } } } ]
Copy code
Common pitfalls to avoid
- Don’t put real secrets in `job_variables` or Prefect Variables; use K8s Secrets or Prefect Secret blocks.
- Don’t rely on manual edits in the UI. Keep the template in Git and update the pool:
prefect work-pool update my-k8s-pool --base-job-template ./base-job-template.json ``` - Block/Variable placeholders like 
{{ prefect.blocks.secret... }}
only resolve in the base job template, not inside 
job_variables
. Docs - Work pools: Work pools - Kubernetes worker: Kubernetes worker - Blocks: Blocks - Variables: Variables If you can share your current base job template or how you’re entering env vars today, I can tailor this to your setup (e.g., convert manual entries into a safe, version-controlled template).
v
@Marvin how to use config map to inject env variables for the workpool?
m
✅ thought for 42.0 seconds
You’ve got two solid ways to use a ConfigMap to inject env vars for a Kubernetes work pool in Prefect 3.x: A) Bulk import everything from a ConfigMap with envFrom B) Pull specific keys with valueFrom.configMapKeyRef The right place to wire this up is your work pool’s base job template so it’s versioned and not edited by hand. Then use deployment 
job_variables
only for non-sensitive overrides. Step 1: Create your ConfigMap in the same namespace your jobs run
Copy code
# Example: app-config with a few keys
kubectl create configmap app-config \
  --from-literal=LOG_LEVEL=INFO \
  --from-literal=TIMEOUT=30 \
  -n prefect
Step 2: Add ConfigMap injection to the base job template Option A: envFrom (bring all keys in as env variables)
Copy code
{
  "job_configuration": {
    "namespace": "{{ namespace }}"
  },
  "job_manifest": {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
      "labels": "{{ labels }}",
      "namespace": "{{ namespace }}",
      "generateName": "{{ name }}-"
    },
    "spec": {
      "backoffLimit": "{{ backoff_limit }}",
      "template": {
        "spec": {
          "restartPolicy": "Never",
          "serviceAccountName": "{{ service_account_name }}",
          "containers": [
            {
              "name": "prefect-job",
              "image": "{{ image }}",
              "imagePullPolicy": "{{ image_pull_policy }}",
              "args": "{{ command }}",
              "env": "{{ env }}",
              "envFrom": [
                {
                  "configMapRef": {
                    "name": "{{ config_map_name }}"
                  }
                }
              ]
            }
          ]
        }
      }
    }
  },
  "variables": {
    "type": "object",
    "properties": {
      "namespace": {
        "type": "string",
        "default": "prefect"
      },
      "image": {
        "type": "string",
        "default": "prefecthq/prefect:3-latest"
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "service_account_name": {
        "type": "string",
        "default": "prefect-worker"
      },
      "backoff_limit": {
        "type": "integer",
        "default": 0
      },
      "config_map_name": {
        "type": "string",
        "default": "app-config"
      },
      "env": {
        "type": [
          "object",
          "array",
          "null"
        ],
        "default": null
      }
    }
  }
}
Option B: valueFrom.configMapKeyRef (use specific keys, keep control)
Copy code
"containers": [
  {
    "name": "prefect-job",
    "image": "{{ image }}",
    "args": "{{ command }}",
    "env": [
      {
        "name": "LOG_LEVEL",
        "valueFrom": {
          "configMapKeyRef": {
            "name": "{{ config_map_name }}",
            "key": "LOG_LEVEL"
          }
        }
      },
      {
        "name": "TIMEOUT",
        "valueFrom": {
          "configMapKeyRef": {
            "name": "{{ config_map_name }}",
            "key": "TIMEOUT"
          }
        }
      }
    ]
  }
]
Notes: - Keep 
env": "{{ env }}"
in the template if you want to allow deployments to add/override non-sensitive env via 
job_variables.env
. The worker will merge those with what’s in the template. 
envFrom
entries are preserved as-is. - Make sure the ConfigMap exists in the same namespace as 
{{ namespace }}
. Step 3: Apply the template to your work pool
Copy code
# Create (or update) a Kubernetes work pool with the template
prefect work-pool create my-k8s-pool --type kubernetes --base-job-template ./base-job-template.json
# or
prefect work-pool update my-k8s-pool --base-job-template ./base-job-template.json
Step 4: Override per deployment (non-sensitive only) ``` from prefect import flow, serve @flow def my_flow(): ... if name == "__main__": serve( my_flow.to_deployment( name="staging", work_pool_name="my-k8s-pool", job_variables={ "namespace": "prefect", "image": "ghcr.io/acme/app:1.2.3", "config_map_name": "app-config", # optional override per env "env": {
"EXTRA_FLAG": "true" # non-sensitive override } } ) ) ``` Best practices - Keep ConfigMap references in the work pool template (version-controlled), not manually typed in the UI. - Avoid duplicate variable names between 
envFrom
and explicit 
env
entries; prefer unique names or rely on explicit 
env
to intentionally override. - Use ConfigMaps for non-sensitive config only. For secrets, use Kubernetes Secrets (
envFrom.secretRef
or 
valueFrom.secretKeyRef
) or Prefect Secret blocks in the template. - Ensure the ConfigMap is present in the job’s namespace before runs. Docs - Work pools: Work pools - Kubernetes worker: Kubernetes worker If you share your current template, I can drop in the exact 
envFrom
or 
configMapKeyRef
wiring and variable schema for you.
v
@Marvin this is how I'm creating workpool async with get_client() as client: work_pool_filter = WorkPoolFilter(name=WorkPoolFilterName(any_=[name])) work_pools = await client.read_work_pools(work_pool_filter=work_pool_filter) # if work pool exist return here if work_pools: return work_pools[0] payload = WorkPoolCreate(name=name, type=type, is_paused=False) if type == PrefectWorker.KUBERNETES.value: payload.base_job_template = k8_base_job_template base = payload.base_job_template if base: props = base.setdefault("variables", {}).setdefault( "properties", {} ) base.setdefault("job_configuration", {}) if namespace: props.setdefault("namespace", {"type": "string"}) props["namespace"]["default"] = namespace base["job_configuration"]["namespace"] = "{{ namespace }}" if image_identifier and image_pull_secret: props.setdefault("image", {"type": "string"}) props["image"]["default"] = image_identifier base["job_configuration"]["image"] = "{{ image }}" if not settings.is_development_mode: props.setdefault("image_pull_secret", {"type": "string"}) props["image_pull_secret"]["default"] = image_pull_secret base["job_configuration"][ "image_pull_secret" ] = "{{ image_pull_secret }}" payload.base_job_template = base # create a new work pool worker_pool = await client.create_work_pool( work_pool=payload, ) how can I inject all environment variables into workpool ? give me step by step guide if all the environment variables will automatically loaded into workpool with their respective names in config map? If I want to rename them the each with different name how can I do that ? Give me step by step guide to fix this
m
✅ thought for 119.7 seconds
Thanks for sharing your pool-creation code — that helps a lot. Below is a step-by-step way to wire ConfigMaps into your Kubernetes work pool so pods get env vars reliably, plus how to “rename” keys. Key concepts up front - “Inject into work pool” means: add references in the work pool’s base job template so the worker injects env vars into each Job at runtime. The values themselves live in Kubernetes (ConfigMap/Secret) or Prefect (Blocks/Variables). - To load ALL keys from a ConfigMap as env vars, use envFrom.configMapRef. Keys become env var names as-is. - To rename keys, you must use per-key valueFrom.configMapKeyRef entries. Kubernetes cannot rename keys when using envFrom. Step-by-step: load all keys from a ConfigMap (envFrom) 1) Create your ConfigMap in the target namespace
Copy code
kubectl create configmap app-config \
  --from-literal=LOG_LEVEL=INFO \
  --from-literal=TIMEOUT=30 \
  -n <your-namespace>
2) Ensure your base job template supports a config_map_name variable and merging of extra env - Add variables to your template schema: - config_map_name (string) - env (object or array) so deployments can still add non-sensitive env - Add envFrom with a reference to that variable in the Job manifest - Keep env merge capability by including env: "{{ env }}" OR rely on the worker merging (details in Option A/B below) Minimal template shape to add (JSON you embed in your 
k8_base_job_template
)
Copy code
{
  "variables": {
    "type": "object",
    "properties": {
      "config_map_name": {
        "type": "string",
        "default": "app-config"
      },
      "env": {
        "type": [
          "object",
          "array",
          "null"
        ],
        "default": null
      },
      "namespace": {
        "type": "string",
        "default": "default"
      },
      "image": {
        "type": "string",
        "default": "prefecthq/prefect:3-latest"
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "service_account_name": {
        "type": "string",
        "default": "prefect-worker"
      },
      "backoff_limit": {
        "type": "integer",
        "default": 0
      }
    }
  },
  "job_configuration": {
    "namespace": "{{ namespace }}"
  },
  "job_manifest": {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
      "labels": "{{ labels }}",
      "namespace": "{{ namespace }}",
      "generateName": "{{ name }}-"
    },
    "spec": {
      "backoffLimit": "{{ backoff_limit }}",
      "template": {
        "spec": {
          "restartPolicy": "Never",
          "serviceAccountName": "{{ service_account_name }}",
          "containers": [
            {
              "name": "prefect-job",
              "image": "{{ image }}",
              "imagePullPolicy": "{{ image_pull_policy }}",
              "args": "{{ command }}",
              "env": "{{ env }}",
              "envFrom": [
                {
                  "configMapRef": {
                    "name": "{{ config_map_name }}"
                  }
                }
              ]
            }
          ]
        }
      }
    }
  }
}
3) Update your Python that builds the work pool payload In your posted code, extend the 
base_job_template
to define 
config_map_name
and add the envFrom block if it isn’t already present. ``` base = payload.base_job_template or {} props = base.setdefault("variables", {}).setdefault("properties", {}) job_conf = base.setdefault("job_configuration", {}) job_manifest = base.setdefault("job_manifest", {}) # Variables (schema) props.setdefault("env", {"type": ["object", "array", "null"], "default": None}) props.setdefault("config_map_name", {"type": "string"}) if not props["config_map_name"].get("default"): props["config_map_name"]["default"] = "app-config" # Ensure manifest structure exists pod_spec = ( job_manifest .setdefault("spec", {}) .setdefault("template", {}) .setdefault("spec", {}) ) containers = pod_spec.setdefault("containers", [{"name": "prefect-job"}]) container = containers[0] container.setdefault("image", "{{ image }}") container.setdefault("imagePullPolicy", "{{ image_pull_policy }}")
container.setdefault("args", "{{ command }}") # Allow deployments to add env vars via job_variables.env container.setdefault("env", "{{ env }}") # Add envFrom for the ConfigMap env_from = container.setdefault("envFrom", []) # Avoid duplicates if run multiple times if not any("configMapRef" in e and e["configMapRef"].get("name") == "{{ config_map_name }}" for e in env_from): env_from.append({"configMapRef": {"name": "{{ config_map_name }}"}}) payload.base_job_template = base
Copy code
4) Result at runtime
- All keys in the ConfigMap will be available as env vars in the pod with the same names.
- You can still add non-sensitive overrides at deployment via `job_variables.env` (dict), which Prefect merges.

Answering your questions
- “Will all environment variables automatically load with their names from the ConfigMap?”
  - Yes, if you use envFrom.configMapRef as above, all keys from the ConfigMap become env vars with the same names in the container. They are not “stored in the work pool”; they’re injected into each Job at runtime.
- “How do I rename each env var to a different name?”
  - You cannot rename with envFrom. Use explicit per-key entries with valueFrom.configMapKeyRef and set the env var name you want. See below.

Step-by-step: rename keys using valueFrom.configMapKeyRef
Option A — bake renames into the template (static list)
- Replace the container’s env value with a list of entries that reference your ConfigMap and keys; set “name” to the new env var name.
- Prefect will still merge deployment `job_variables.env` on top of this list at runtime.

Snippet to alter your `base_job_template` in code:
# Replace or augment container env with a list that includes valueFrom entries env_list = container.setdefault("env", []) if isinstance(env_list, str): # if it was "{{ env }}", convert to list env_list = [] container["env"] = env_list # Add remapped keys (example) def add_env_from_cm(target_name, cm_name_var, cm_key): entry = { "name": target_name, "valueFrom": { "configMapKeyRef": { "name": f"{{{{ {cm_name_var} }}}}", # "{{ config_map_name }}" "key": cm_key } } } # avoid duplicates if not any(e.get("name") == target_name for e in env_list): env_list.append(entry) props.setdefault("config_map_name", {"type": "string", "default": "app-config"}) add_env_from_cm("APP_LOG_LEVEL", "config_map_name", "LOG_LEVEL") add_env_from_cm("REQUEST_TIMEOUT", "config_map_name", "TIMEOUT")
Copy code
Option B — allow per-deployment renames without changing the template
- Keep the template with `envFrom` OR no env at all.
- Define `env` in variables schema as `["object", "array"]`.
- In a deployment, pass `job_variables.env` as a list of Kubernetes env entries with valueFrom. The worker preserves these.

Example deployment override:
work_pool_name="my-k8s-pool", job_variables={ "env": [ { "name": "APP_LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "app-config", "key": "LOG_LEVEL" } } }, { "name": "REQUEST_TIMEOUT", "valueFrom": { "configMapKeyRef": { "name": "app-config", "key": "TIMEOUT" } } } ] } ``` Notes: - Because 
env
allows an array, you can pass valueFrom entries from deployments. Prefect merges these with any env/envFrom in the template. - This is great when each environment needs a slightly different mapping without changing the pool template. What I recommend - If you just need “all keys, same names”: use envFrom.configMapRef in the base template (simplest). - If you need renames: either - Hardcode a small, stable list of renames in the template using valueFrom.configMapKeyRef; or - Keep the template generic and pass a list of valueFrom entries per deployment via 
job_variables.env
(most flexible).
- Continue to use 
job_variables.env
(dict) for non-sensitive per-deployment flags; use ConfigMaps/Secrets for values managed by K8s. Docs - Work pools: Work pools - Kubernetes worker: Kubernetes worker If you paste your current 
k8_base_job_template
dict, I can give you the exact patch to add 
envFrom
and (optionally) the rename list so it fits your structure.
v
@Marvin Give me clear instructions to fix this .. the above looks clumsy and it doesn't follow any step by step or clean explanation to do it
m
✅ thought for 68.0 seconds
Absolutely — here’s a clean, step-by-step guide you can drop into your codebase. It covers both “load all keys as-is” and “rename keys” options. Assumptions: - Prefect 3.x - You’re creating the pool via the Python client (as in your snippet) - You want to source env vars from a Kubernetes ConfigMap Step 0 — Decide your mode - Load all keys from a ConfigMap with the same names → use envFrom.configMapRef (simplest) - Rename keys (map ConfigMap keys to different env var names) → use valueFrom.configMapKeyRef entries Step 1 — Create the ConfigMap (once per namespace)
Copy code
kubectl create configmap app-config \
  --from-literal=LOG_LEVEL=INFO \
  --from-literal=TIMEOUT=30 \
  -n <your-namespace>
Step 2A — Load ALL keys as-is (envFrom.configMapRef) Add these changes to the work pool’s base job template before calling create_work_pool. This keeps env from deployments working too. Drop-in helper:
Copy code
def add_configmap_env_from(base, default_config_map_name="app-config"):
    props = base.setdefault("variables", {}).setdefault("properties", {})
    props.setdefault("config_map_name", {"type": "string", "default": default_config_map_name})
    props.setdefault("env", {"type": ["object", "array", "null"], "default": None})

    jm = base.setdefault("job_manifest", {})
    tmpl_spec = jm.setdefault("spec", {}).setdefault("template", {}).setdefault("spec", {})
    containers = tmpl_spec.setdefault("containers", [{"name": "prefect-job"}])
    c = containers[0]
    c.setdefault("image", "{{ image }}")
    c.setdefault("imagePullPolicy", "{{ image_pull_policy }}")
    c.setdefault("args", "{{ command }}")

    # Allow deployments to add/override env with job_variables.env
    c.setdefault("env", "{{ env }}")

    # Ensure envFrom includes our ConfigMap
    env_from = c.setdefault("envFrom", [])
    if not any(e.get("configMapRef", {}).get("name") == "{{ config_map_name }}" for e in env_from):
        env_from.append({"configMapRef": {"name": "{{ config_map_name }}"}})
    return base
Use it in your existing code (inside your KUBERNETES block):
Copy code
payload.base_job_template = k8_base_job_template or {}
base = payload.base_job_template

# your existing namespace/image setup here...

base = add_configmap_env_from(base, default_config_map_name="app-config")
payload.base_job_template = base
Behavior: - All keys from ConfigMap become env vars with identical names in every Job. - You can still pass non-sensitive env via deployment 
job_variables.env
(dict). Prefect merges them at runtime. Step 2B — Rename keys (valueFrom.configMapKeyRef) If you need different env var names than the keys in the ConfigMap, build explicit entries. This still lets deployment env merge on top. Drop-in helper: ``` def add_configmap_key_renames(base, mapping, default_config_map_name="app-config"): # mapping example: {"APP_LOG_LEVEL": "LOG_LEVEL", "REQUEST_TIMEOUT": "TIMEOUT"} props = base.setdefault("variables", {}).setdefault("properties", {}) props.setdefault("config_map_name", {"type": "string", "default": default_config_map_name}) props.setdefault("env", {"type": ["object", "array", "null"], "default": None}) jm = base.setdefault("job_manifest", {}) tmpl_spec = jm.setdefault("spec", {}).setdefault("template", {}).setdefault("spec", {}) containers = tmpl_spec.setdefault("containers", [{"name": "prefect-job"}]) c = containers[0] c.setdefault("image", "{{ image }}") c.setdefault("imagePullPolicy", "{{ image_pull_policy }}") c.setdefault("args", "{{ command }}") # Use a list so the worker can merge deployment env on top if c.get("env") == "{{ env }}": c["env"] = [] c.setdefault("env", []) existing = {e.get("name") for e in c["env"] if isinstance(e, dict)} for new_name, cm_key in mapping.items(): if new_name in existing: continue c["env"].append({ "name": new_name, "valueFrom": { "configMapKeyRef": {
"name": "{{ config_map_name }}", "key": cm_key } } }) return base
Copy code
Use it in your existing code:
payload.base_job_template = k8_base_job_template or {} base = payload.base_job_template # your existing namespace/image setup here... rename_map = { "APP_LOG_LEVEL": "LOG_LEVEL", "REQUEST_TIMEOUT": "TIMEOUT", } base = add_configmap_key_renames(base, rename_map, default_config_map_name="app-config") payload.base_job_template = base
Copy code
Behavior:
- Each Job gets env vars named APP_LOG_LEVEL and REQUEST_TIMEOUT, whose values come from LOG_LEVEL and TIMEOUT keys in the ConfigMap.
- Deployment `job_variables.env` (dict) still merges on top at runtime; the worker converts dict to list and merges with your explicit list.

Step 3 — Keep your existing namespace/image logic
Your snippet adding namespace/image defaults is fine. Keep it as-is alongside Step 2A or 2B.

Step 4 — Create the pool (your code already does this)
You’re already calling:
worker_pool = await client.create_work_pool(work_pool=payload)
Copy code
Step 5 — Per-deployment overrides (optional, non-sensitive)
Example to add/override non-sensitive env in a deployment:
job_variables={ "config_map_name": "app-config", # optional, if you want to override per env "env": { # these will merge on top at runtime "EXTRA_FLAG": "true", "LOG_LEVEL": "DEBUG" # will override if same var name exists } }
Copy code
Answers to your specific questions
- “Will all the environment variables automatically load into the work pool with their names from the ConfigMap?”
  - Yes, if you use envFrom.configMapRef (Step 2A), all keys are injected as environment variables with the same names in each Job.
- “If I want to rename them with different names, how can I do that?”
  - Use valueFrom.configMapKeyRef entries (Step 2B). You define the env var name you want, and point to the original ConfigMap key.

Quick test
- Add a tiny flow that prints env:
import os from prefect import flow @flow def show_env(): print("LOG_LEVEL:", os.getenv("LOG_LEVEL")) print("APP_LOG_LEVEL:", os.getenv("APP_LOG_LEVEL")) ``` - Deploy to your pool and run once; confirm the values are present as expected based on Step 2A or 2B. Notes - Use ConfigMaps for non-sensitive config. For secrets, use Kubernetes Secrets (
envFrom.secretRef
or 
valueFrom.secretKeyRef
) or Prefect Secret blocks in the base template. - The worker merges deployment 
job_variables.env
on top of what’s in the template; duplicates are overridden by the deployment. If you paste your current 
k8_base_job_template
dict, I can return an exact patched version for Step 2A or 2B so you can drop it in without guesswork. Docs: - Work pools: Work pools - Kubernetes worker: Kubernetes worker
v
@Marvin I've just added this and method base = self.add_configmap_env_from( base, default_config_map_name="config-map" ) Not used the rename to check if all envs are coming inside flow it didn't show any evn in the running flow
m
✅ thought for 85.4 seconds
Thanks for the clarification — let’s make this crisp and actionable. From your description, the most likely reason you’re not seeing env vars in the flow is that your code returns early when the pool already exists, so the updated base job template (with envFrom) is never applied. Second most common issue is a namespace or ConfigMap name mismatch. Follow these steps exactly: 1) Verify your ConfigMap exists in the job’s namespace - Whatever namespace your jobs run in (your 
{{ namespace }}
default or override), create/verify the ConfigMap there.
Copy code
# Replace <ns> with the namespace your jobs run in
kubectl -n <ns> get configmap
kubectl -n <ns> describe configmap config-map
- If it’s missing, create it:
Copy code
kubectl -n <ns> create configmap config-map \
  --from-literal=LOG_LEVEL=INFO \
  --from-literal=TIMEOUT=30
2) Ensure your work pool actually has envFrom in its base job template - Your helper is fine (envFrom is the right approach), but you must update the pool even if it exists. Right now your code returns early. Modify your code so it updates existing pools:
Copy code
async with get_client() as client:
    work_pool_filter = WorkPoolFilter(name=WorkPoolFilterName(any_=[name]))
    work_pools = await client.read_work_pools(work_pool_filter=work_pool_filter)

    # Always build/patch the base template in memory
    base = k8_base_job_template or {}
    # Your existing code that sets namespace/image defaults...
    # Then add envFrom (your helper)
    base = self.add_configmap_env_from(base, default_config_map_name="config-map")

    if work_pools:
        # UPDATE the existing pool with the new base job template
        await client.update_work_pool(
            work_pool_name=name,
            base_job_template=base
        )
        pool = await client.read_work_pool(work_pool_name=name)
        return pool

    # CREATE a new pool with the base job template
    payload = WorkPoolCreate(name=name, type=type, is_paused=False)
    if type == PrefectWorker.KUBERNETES.value:
        payload.base_job_template = base
    return await client.create_work_pool(work_pool=payload)
Notes: - The important change is to call 
update_work_pool
when the pool already exists. Otherwise your new envFrom never takes effect. 3) Confirm the pool template contains envFrom - After running the code above, inspect the pool:
Copy code
prefect work-pool inspect "<your-pool-name>" --output json
- Check that it includes:
Copy code
... job_manifest.spec.template.spec.containers[0].envFrom[0].configMapRef.name == "{{ config_map_name }}" ...
- Also confirm variables include 
config_map_name
with default “config-map”. 4) Re-run a flow and verify from the pod Sometimes it’s easier to validate from Kubernetes directly: - Trigger a flow run using this pool. - Find the pod:
Copy code
kubectl -n <ns> get pods -l job-name=<prefect-job-name>
- Exec into the container and print env:
Copy code
kubectl -n <ns> exec -it <pod-name> -- env | egrep 'LOG_LEVEL|TIMEOUT'
You should see the values from the ConfigMap. If you don’t: - Double-check the pod’s namespace and the ConfigMap name match. - Make sure your pool template truly has envFrom (Step 3). 5) If you still want to print from the flow - Add a quick check in your flow:
Copy code
import os
from prefect import flow

@flow
def show_env():
    print("LOG_LEVEL:", os.getenv("LOG_LEVEL"))
    print("TIMEOUT:", os.getenv("TIMEOUT"))
- Deploy/run with the updated pool. If it prints None, it usually means the pod didn’t get envFrom (template wasn’t updated) or the ConfigMap/namespace didn’t match. 6) Optional: per-deployment overrides still work - Keep 
env": "{{ env }}"
in the template so you can add non-sensitive env via 
job_variables.env
. These are merged on top at runtime:
Copy code
job_variables={
  "config_map_name": "config-map",  # override if needed
  "env": { "EXTRA_FLAG": "true" }
}
7) If you need renaming instead of loading all keys
- envFrom cannot rename. Use explicit entries with valueFrom.configMapKeyRef in the base template or pass them per deployment as a list:
Copy code
job_variables={
  "env": [
    {
      "name": "APP_LOG_LEVEL",  # new env var name
      "valueFrom": {
        "configMapKeyRef": { "name": "config-map", "key": "LOG_LEVEL" }
      }
    }
  ]
}
Common pitfalls to check - Pool not updated because your code returned early (fix in Step 2). - ConfigMap exists in a different namespace than the job (Step 1). - Using a different pool or worker type than you think (ensure deployment uses this kubernetes pool and a Kubernetes worker is running). - Inspecting only Prefect-managed env in the UI; envFrom vars won’t show there but will exist in the pod. Docs - Work pools: Work pools - Kubernetes worker: Kubernetes worker If you paste the result of 
prefect work-pool inspect "<pool>" --output json
(or your current 
base_job_template
dict), I’ll pinpoint exactly what’s missing.
v
@Marvin Here is my k8s base job template do I need to fix anything { "variables": { "type": "object", "properties": { "env": { "type": "object", "title": "Environment Variables", "description": "Environment variables to set when starting a flow run.", "additionalProperties": { "anyOf": [ { "type": "string" }, { "type": "null" } ] } }, "name": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Name", "description": "Name given to infrastructure created by a worker." }, "image": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Image", "examples": [ "docker.io/prefecthq/prefect:3-latest" ], "description": "The image reference of a container image to use for created jobs. If not set, the latest Prefect image will be used." }, "labels": { "type": "object", "title": "Labels", "description": "Labels applied to infrastructure created by a worker.", "additionalProperties": { "type": "string" } }, "command": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Command", "description": "The command to use when starting a flow run. In most cases, this should be left blank and the command will be automatically generated by the worker." }, "namespace": { "type": "string", "title": "Namespace", "description": "The Kubernetes namespace to create jobs within." }, "stream_output": { "type": "boolean", "title": "Stream Output", "default": true, "description": "If set, output will be streamed from the job to local standard output." }, "cluster_config": { "anyOf": [ { "$ref": "#/definitions/KubernetesClusterConfig" }, { "type": "null" } ], "description": "The Kubernetes cluster config to use for job creation." }, "finished_job_ttl": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "title": "Finished Job TTL", "default": 0, "description": "The number of seconds to retain jobs after completion. If set, finished jobs will be cleaned up by Kubernetes after the given delay. If not set, jobs will be retained indefinitely." }, "image_pull_policy": { "enum": [ "IfNotPresent", "Always", "Never" ], "type": "string", "title": "Image Pull Policy", "default": "IfNotPresent", "description": "The Kubernetes image pull policy to use for job containers." }, "image_pull_secret": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Image pull secret", "description": "Image pull secret." }, "resource_limits_cpu": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Resource Limits - CPU (millicores)", "description": "Limiting by CPU." }, "service_account_name": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Service Account Name", "description": "The Kubernetes service account to use for job creation." }, "resource_requests_cpu": { "anyOf": [ { "type": "string" }, { "type": "null" } ], "title": "Resource Requests - CPU (millicores)", "description": "Requested CPU." }, "resource_limits_memory": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "title": "Resource Limits - Memory (MB)", "description": "Limiting by Memory." }, "resource_requests_memory": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "title": "Resource Requests - Memory (MB)", "description": "Requested Memory." }, "job_watch_timeout_seconds": { "anyOf": [ { "type": "integer" }, { "type": "null" } ], "title": "Job Watch Timeout Seconds", "description": "Number of seconds to wait for each event emitted by a job before timing out. If not set, the worker will wait for each event indefinitely." }, "pod_watch_timeout_seconds": { "type": "integer", "title": "Pod Watch Timeout Seconds", "default": 60, "description": "Number of seconds to watch for pod creation before timing out." } }, "definitions": { "KubernetesClusterConfig": { "type": "object", "title": "KubernetesClusterConfig", "required": [ "config", "context_name" ], "properties": { "config": { "type": "object", "title": "Config", "description": "The entire contents of a kubectl config file." }, "context_name": { "type": "string", "title": "Context Name", "description": "The name of the kubectl context to use." } }, "description": "Stores configuration for interaction with Kubernetes clusters.\n\nSee 
from_file
for creation.", "secret_fields": [], "block_type_slug": "kubernetes-cluster-config", "block_schema_references": {} } }, "description": "Default variables for the Kubernetes worker.\n\nThe schema for this class is used to populate the 
variables
section of the default\nbase job template." }, "job_configuration": { "env": "{{ env }}", "name": "{{ name }}", "labels": "{{ labels }}", "command": "{{ command }}", "namespace": "{{ namespace }}", "job_manifest": { "kind": "Job", "spec": { "template": { "spec": { "containers": [ { "env": "{{ env }}", "args": "{{ command }}", "name": "prefect-job", "image": "{{ image }}", "resources": { "limits": { "cpu": "{{ resource_limits_cpu }}m", "memory": "{{ resource_limits_memory }}Mi" }, "requests": { "cpu": "{{ resource_requests_cpu }}m", "memory": "{{ resource_requests_memory }}Mi" } }, "imagePullPolicy": "{{ image_pull_policy }}" } ], "completions": 1, "parallelism": 1, "restartPolicy": "Never", "imagePullSecrets": [ { "name": "{{ image_pull_secret }}" } ], "serviceAccountName": "{{ service_account_name }}" } }, "backoffLimit": 0, "ttlSecondsAfterFinished": "{{ finished_job_ttl }}" }, "metadata": { "labels": "{{ labels }}", "namespace": "{{ namespace }}", "generateName": "{{ name }}-" }, "apiVersion": "batch/v1" }, "stream_output": "{{ stream_output }}", "cluster_config": "{{ cluster_config }}", "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}", "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}" } } Also tell me what should I do to see my vairbales in workper
m
Your message was too long, here's your message at the allowed limit: Here is my k8s base job template do I need to fix anything { variables { type object properties { env { type object title Environment Variables description Environment variables to set when starting a flow run additionalProperties { anyOf [ { type string } { type null } ] } } name { anyOf [ { type string } { type null } ] title Name description Name given to infrastructure created by a worker } image { anyOf [ { type string } { type null } ] title Image examples [ <http //docker io/prefecthq/prefect 3-latest|docker io/prefecthq/prefect 3-latest> ] description The image reference of a container image to use for created jobs If not set the latest Prefect image will be used } labels { type object title Labels description Labels applied to infrastructure created by a worker additionalProperties { type string } } command { anyOf [ { type string } { type null } ] title Command description The command to use when starting a flow run In most cases this should be left blank and the command will be automatically generated by the worker } namespace { type string title Namespace description The Kubernetes namespace to create jobs within } stream_output { type boolean title Stream Output default true description If set output will be streamed from the job to local standard output } cluster_config { anyOf [ { $ref #/definitions/KubernetesClusterConfig } { type null } ] description The Kubernetes cluster config to use for job creation } finished_job_ttl { anyOf [ { type integer } { type null } ] title Finished Job TTL default 0 description The number of seconds to retain jobs after completion If set finished jobs will be cleaned up by Kubernetes after the given delay If not set jobs will be retained indefinitely } image_pull_policy { enum [ IfNotPresent Always Never ] type string title Image Pull Policy default IfNotPresent description The Kubernetes image pull policy to use for job containers } image_pull_secret { anyOf [ { type string } { type null } ] title Image pull secret description Image pull secret } resource_limits_cpu { anyOf [ { type string } { type null } ] title Resource Limits - CPU (millicores) description Limiting by CPU } service_account_name { anyOf [ { type string } { type null } ] title Service Account Name description The Kubernetes service account to use for job creation } resource_requests_cpu { anyOf [ { type string } { type null } ] title Resource Requests - CPU (millicores) description Requested CPU } resource_limits_memory { anyOf [ { type integer } { type null } ] title Resource Limits - Memory (MB) description Limiting by Memory } resource_requests_memory { anyOf [ { type integer } { type null } ] title Resource Requests - Memory (MB) description Requested Memory } job_watch_timeout_seconds { anyOf [ { type integer } { type null } ] title Job Watch Timeout Seconds description Number of seconds to
v
@Marvin this my base job temaplte "job_configuration": { "env": "{{ env }}", "name": "{{ name }}", "labels": "{{ labels }}", "command": "{{ command }}", "namespace": "{{ namespace }}", "job_manifest": { "kind": "Job", "spec": { "template": { "spec": { "containers": [ { "env": "{{ env }}", "args": "{{ command }}", "name": "prefect-job", "image": "{{ image }}", "resources": { "limits": { "cpu": "{{ resource_limits_cpu }}m", "memory": "{{ resource_limits_memory }}Mi" }, "requests": { "cpu": "{{ resource_requests_cpu }}m", "memory": "{{ resource_requests_memory }}Mi" } }, "imagePullPolicy": "{{ image_pull_policy }}" } ], "completions": 1, "parallelism": 1, "restartPolicy": "Never", "imagePullSecrets": [ { "name": "{{ image_pull_secret }}" } ], "serviceAccountName": "{{ service_account_name }}" } }, "backoffLimit": 0, "ttlSecondsAfterFinished": "{{ finished_job_ttl }}" }, "metadata": { "labels": "{{ labels }}", "namespace": "{{ namespace }}", "generateName": "{{ name }}-" }, "apiVersion": "batch/v1" }, "stream_output": "{{ stream_output }}", "cluster_config": "{{ cluster_config }}", "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}", "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}" } check and tell me
m
✅ thought for 96.8 seconds
Thanks for sharing your template. Two issues to fix first, then add the ConfigMap wiring: - Your 
job_manifest
is nested inside 
job_configuration
. In Prefect 3.x, 
job_manifest
must be a top-level sibling of 
job_configuration
. - There’s no 
envFrom
in the container spec, so the ConfigMap is never injected. Follow these steps. Step 1) Fix your base job template structure and add ConfigMap injection - Move 
job_manifest
to the top level - Add a 
config_map_name
variable in the schema - Add 
envFrom.configMapRef
to the container - Keep 
env: "{{ env }}"
so deployment-level 
job_variables.env
can still merge in Use this corrected shape (adapt to your existing variables block; I show only the relevant parts):
Copy code
{
  "variables": {
    "type": "object",
    "properties": {
      "namespace": {
        "type": "string",
        "default": "default"
      },
      "image": {
        "type": "string"
      },
      "image_pull_policy": {
        "type": "string",
        "enum": [
          "IfNotPresent",
          "Always",
          "Never"
        ],
        "default": "IfNotPresent"
      },
      "service_account_name": {
        "type": "string"
      },
      "image_pull_secret": {
        "type": "string"
      },
      "finished_job_ttl": {
        "type": [
          "integer",
          "null"
        ],
        "default": null
      },
      "stream_output": {
        "type": "boolean",
        "default": true
      },
      "cluster_config": {
        "type": [
          "object",
          "null"
        ],
        "default": null
      },
      "job_watch_timeout_seconds": {
        "type": [
          "integer",
          "null"
        ],
        "default": null
      },
      "pod_watch_timeout_seconds": {
        "type": "integer",
        "default": 60
      },
      "resource_limits_cpu": {
        "type": "integer"
      },
      "resource_limits_memory": {
        "type": "integer"
      },
      "resource_requests_cpu": {
        "type": "integer"
      },
      "resource_requests_memory": {
        "type": "integer"
      },
      "config_map_name": {
        "type": "string",
        "default": "config-map"
      },
      "env": {
        "type": [
          "object",
          "array",
          "null"
        ],
        "default": null
      }
    }
  },
  "job_configuration": {
    "env": "{{ env }}",
    "name": "{{ name }}",
    "labels": "{{ labels }}",
    "command": "{{ command }}",
    "namespace": "{{ namespace }}",
    "stream_output": "{{ stream_output }}",
    "cluster_config": "{{ cluster_config }}",
    "job_watch_timeout_seconds": "{{ job_watch_timeout_seconds }}",
    "pod_watch_timeout_seconds": "{{ pod_watch_timeout_seconds }}"
  },
  "job_manifest": {
    "apiVersion": "batch/v1",
    "kind": "Job",
    "metadata": {
      "labels": "{{ labels }}",
      "namespace": "{{ namespace }}",
      "generateName": "{{ name }}-"
    },
    "spec": {
      "backoffLimit": 0,
      "ttlSecondsAfterFinished": "{{ finished_job_ttl }}",
      "template": {
        "spec": {
          "restartPolicy": "Never",
          "serviceAccountName": "{{ service_account_name }}",
          "imagePullSecrets": [
            {
              "name": "{{ image_pull_secret }}"
            }
          ],
          "completions": 1,
          "parallelism": 1,
          "containers": [
            {
              "name": "prefect-job",
              "image": "{{ image }}",
              "imagePullPolicy": "{{ image_pull_policy }}",
              "args": "{{ command }}",
              "env": "{{ env }}",
              "envFrom": [
                {
                  "configMapRef": {
                    "name": "{{ config_map_name }}"
                  }
                }
              ],
              "resources": {
                "limits": {
                  "cpu": "{{ resource_limits_cpu }}m",
                  "memory": "{{ resource_limits_memory }}Mi"
                },
                "requests": {
                  "cpu": "{{ resource_requests_cpu }}m",
                  "memory": "{{ resource_requests_memory }}Mi"
                }
              }
            }
          ]
        }
      }
    }
  }
}
Step 2) Update the existing work pool with this template Your code currently returns early if the pool exists, so changes won’t apply. Always push the updated template: ``` async with get_client() as client: # Build/patch the base template first base = k8_base_job_template or {} # ... your existing code setting namespace/image defaults ...
# Ensure base contains the corrected structure above # If pool exists, update it; else create it work_pools = await client.read_work_pools( work_pool_filter=WorkPoolFilter(name=WorkPoolFilterName(any_=[name])) ) if work_pools: await client.update_work_pool( work_pool_name=name, base_job_template=base ) return await client.read_work_pool(work_pool_name=name) payload = WorkPoolCreate(name=name, type=type, is_paused=False) if type == PrefectWorker.KUBERNETES.value: payload.base_job_template = base return await client.create_work_pool(work_pool=payload)
Copy code
Step 3) Verify the pool configuration
prefect work-pool inspect "<your-pool-name>" --output json
Copy code
Confirm:
- `job_manifest` is top-level (sibling of `job_configuration`)
- `containers[0].envFrom[0].configMapRef.name` is `{{ config_map_name }}`
- `variables.properties.config_map_name.default` is `"config-map"`

Step 4) Ensure the ConfigMap exists in the job’s namespace
Replace <ns> with the namespace your jobs use (from `{{ namespace }}`):
kubectl -n <ns> get configmap kubectl -n <ns> describe configmap config-map # If missing, create: kubectl -n <ns> create configmap config-map \ --from-literal=LOG_LEVEL=INFO \ --from-literal=TIMEOUT=30
Copy code
Step 5) Run a flow and verify
- Trigger a run that uses this pool.
- From the pod, check env:
kubectl -n <ns> get pods -l job-name=<prefect-job-name> kubectl -n <ns> exec -it <pod-name> -- env | egrep 'LOG_LEVEL|TIMEOUT'
Copy code
You should see the ConfigMap keys as env vars. If not, re-check Steps 2–4.

Optional: renaming keys
- envFrom cannot rename. To map keys to new names, use explicit entries in the template or per deployment:

Template approach (bake renames):
"env": [ { "name": "APP_LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "{{ config_map_name }}", "key": "LOG_LEVEL" } } }, { "name": "REQUEST_TIMEOUT", "valueFrom": { "configMapKeyRef": { "name": "{{ config_map_name }}", "key": "TIMEOUT" } } } ]
Copy code
Per-deployment approach (flexible):
job_variables={ "env": [ { "name": "APP_LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "config-map", "key": "LOG_LEVEL" } } }, { "name": "REQUEST_TIMEOUT", "valueFrom": { "configMapKeyRef": { "name": "config-map", "key": "TIMEOUT" } } } ] } ``` Common pitfalls - Pool not updated (fix by calling 
update_work_pool
). - ConfigMap in the wrong namespace (must match 
{{ namespace }}
). - Using a different pool/worker than the deployment specifies. - Expecting env vars to appear in Prefect UI; they exist in the pod’s environment, verify via kubectl. Docs - Work pools: Work pools - Kubernetes worker: Kubernetes worker If you paste your full base template JSON (including the 
variables
block) and the output of 
prefect work-pool inspect "<pool>" --output json
, I can validate the exact structure for you.
v
@Marvin i'm using 2X prefect version give me code acorrding to that
m
✅ thought for 143.9 seconds
Got it — here’s a clean, step-by-step 2.x solution with ready-to-paste code. Assumptions - Prefect 2.x - You’re using a Kubernetes work pool and creating/updating it via the Python client (like your snippet) - Goal: inject env vars from a Kubernetes ConfigMap into every job created by this pool High-level - Use envFrom.configMapRef in the pool’s base job template - Ensure the ConfigMap exists in the same namespace your jobs run in - Update the pool’s base job template even if the pool already exists (don’t return early) Step 1) Install the Kubernetes integration (2.x)
Copy code
pip install prefect-kubernetes
Step 2) Create (or verify) the ConfigMap in the job namespace Replace <ns> with the namespace your jobs use (the default/override for 
{{ namespace }}
):
Copy code
kubectl -n <ns> create configmap config-map \
  --from-literal=LOG_LEVEL=INFO \
  --from-literal=TIMEOUT=30
# or verify it exists
kubectl -n <ns> describe configmap config-map
Step 3) Patch your base job template to: - Move job_manifest to the top level (in 2.x, it should not be nested under job_configuration) - Add envFrom.configMapRef - Add variables 
config_map_name
and 
env
- Keep 
env: "{{ env }}"
so per-deployment 
job_variables.env
can still merge in Drop-in helper to fix/augment your template dict (2.x-safe)
Copy code
def ensure_configmap_env_from(base: dict, default_config_map_name: str = "config-map") -> dict:
    # Variables schema
    props = base.setdefault("variables", {}).setdefault("properties", {})
    props.setdefault("config_map_name", {"type": "string", "default": default_config_map_name})
    props.setdefault("env", {"type": ["object", "array", "null"], "default": None})

    # Move job_manifest to top-level if it was nested
    job_conf = base.setdefault("job_configuration", {})
    if "job_manifest" in job_conf and "job_manifest" not in base:
        base["job_manifest"] = job_conf.pop("job_manifest")
    jm = base.setdefault("job_manifest", {})

    # Ensure basic manifest structure exists
    jm.setdefault("apiVersion", "batch/v1")
    jm.setdefault("kind", "Job")
    jm.setdefault("metadata", {}).setdefault("generateName", "{{ name }}-")
    jm["metadata"].setdefault("labels", "{{ labels }}")
    jm["metadata"].setdefault("namespace", "{{ namespace }}")

    spec = jm.setdefault("spec", {})
    spec.setdefault("backoffLimit", 0)
    spec.setdefault("ttlSecondsAfterFinished", "{{ finished_job_ttl }}")
    tmpl = spec.setdefault("template", {})
    pod_spec = tmpl.setdefault("spec", {})

    # Standard pod spec defaults
    pod_spec.setdefault("restartPolicy", "Never")
    pod_spec.setdefault("serviceAccountName", "{{ service_account_name }}")
    pod_spec.setdefault("imagePullSecrets", [{"name": "{{ image_pull_secret }}" }])
    pod_spec.setdefault("completions", 1)
    pod_spec.setdefault("parallelism", 1)

    containers = pod_spec.setdefault("containers", [{"name": "prefect-job"}])
    c = containers[0]
    c.setdefault("name", "prefect-job")
    c.setdefault("image", "{{ image }}")
    c.setdefault("imagePullPolicy", "{{ image_pull_policy }}")
    c.setdefault("args", "{{ command }}")

    # Allow deployment-level env to merge in
    if "env" not in c or not c["env"]:
        c["env"] = "{{ env }}"

    # Add envFrom for the ConfigMap (avoid duplicate)
    env_from = c.setdefault("envFrom", [])
    if not any(e.get("configMapRef", {}).get("name") == "{{ config_map_name }}" for e in env_from):
        env_from.append({"configMapRef": {"name": "{{ config_map_name }}"}})

    return base
Step 4) Update your work-pool creation code to apply the patch AND update existing pools ``` from prefect.client.orchestration import get_client from prefect.server.schemas.actions import WorkPoolCreate from prefect.server.schemas.filters import WorkPoolFilter, WorkPoolFilterName async def upsert_k8s_pool(name: str, pool_type: str, k8_base_job_template: dict): async with get_client() as client:
# Build base template and add ConfigMap wiring base = k8_base_job_template or {} base = ensure_configmap_env_from(base, default_config_map_name="config-map") # Add your existing defaults (namespace/image/etc.) here as you already do... # e.g., base['job_configuration']['namespace'] = "{{ namespace }}" etc. # Check for existing pool pools = await client.read_work_pools( work_pool_filter=WorkPoolFilter(name=WorkPoolFilterName(any_=[name])) ) if pools: # UPDATE existing pool with new base template await client.update_work_pool( work_pool_name=name, base_job_template=base ) return await client.read_work_pool(work_pool_name=name) # CREATE new pool payload = WorkPoolCreate(name=name, type=pool_type, is_paused=False) payload.base_job_template = base return await client.create_work_pool(work_pool=payload)
Copy code
Step 5) Verify the pool actually contains envFrom
prefect work-pool inspect "<your-pool-name>" --output json
Copy code
You should see:
- Top-level `job_manifest`
- `containers[0].envFrom[0].configMapRef.name == "{{ config_map_name }}"`
- `variables.properties.config_map_name.default == "config-map"`

Step 6) Run a test flow and confirm the env is present
A quick flow that prints variables:
import os from prefect import flow @flow def show_env(): print("LOG_LEVEL:", os.getenv("LOG_LEVEL")) print("TIMEOUT:", os.getenv("TIMEOUT"))
Copy code
Deploy/run using this pool. If you don’t see values:
- Confirm the pod’s namespace and ConfigMap name match
- Confirm the pool template was updated (Step 5)
- Exec into the pod to inspect env directly:
kubectl -n <ns> exec -it <pod-name> -- env | egrep 'LOG_LEVEL|TIMEOUT'
Copy code
Optional: Renaming keys (ConfigMap key → different env var name)
envFrom cannot rename; use explicit entries:
- Template-based (bake remaps into the pool)
# In your base template container spec: "env": [ { "name": "APP_LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "{{ config_map_name }}", "key": "LOG_LEVEL" } } }, { "name": "REQUEST_TIMEOUT", "valueFrom": { "configMapKeyRef": { "name": "{{ config_map_name }}", "key": "TIMEOUT" } } } ]
Copy code
- Or pass per deployment as a list in `job_variables.env`:
job_variables={ "env": [ { "name": "APP_LOG_LEVEL", "valueFrom": { "configMapKeyRef": { "name": "config-map", "key": "LOG_LEVEL" } } }, { "name": "REQUEST_TIMEOUT", "valueFrom": { "configMapKeyRef": { "name": "config-map", "key": "TIMEOUT" } } } ] } ``` Notes and gotchas (2.x) - Make sure a Kubernetes worker is running against this pool. - The ConfigMap must exist in the same namespace your jobs run in. - Env vars from envFrom won’t show in Prefect UI; verify in the pod or print from your flow. - This pattern is for non-sensitive config. For secrets, use Kubernetes Secrets (
envFrom.secretRef
or 
valueFrom.secretKeyRef
) or Prefect Secret blocks in the template. If you paste your full base_job_template (including the variables block) and the output of 
prefect work-pool inspect "<pool>" --output json
, I can validate the exact structure for 2.x and return a patched JSON you can drop in.
v
@Marvin How can I create a workpool with name with create workpool not a workpool name insfrature name of the workpool
m
✅ thought for 109.7 seconds
It sounds like you want to name the work pool itself (not the underlying infrastructure). In Prefect 3.x, you name the pool when you create it, then (optionally) provision infrastructure for that pool separately. - Create a work pool with a specific name:
Copy code
prefect work-pool create "my-pool" --type docker
# or
prefect work-pool create "my-pool" --type kubernetes
# or specify another type as needed
- Provision infrastructure for that existing pool (you still pass the pool’s name here):
Copy code
prefect work-pool provision-infrastructure "my-pool"
This provisions infra for the pool; you don’t pass a separate “infrastructure name” to this command. It derives resource names from the pool context and the provisioning flow. - List pools to confirm:
Copy code
prefect work-pool ls
- If you meant “rename a work pool”: there isn’t a direct rename command. Create a new pool with the desired name, update your deployments to point to it, then delete the old pool. Quick check: - If your pool name contains spaces, quote it: 
"My Pool"
. Many users prefer hyphens: 
my-pool
. Can you share which pool type you want (docker, kubernetes, etc.) and the exact name? I can give you the exact command tailored to your setup.
v
@Marvin My workpool name is grammar-processor and my flow name is grammar-processor and my deployment name is grammar-processor-deployment but when I tried to run deplooyment it says read deployment by name failed what could be the problem? do I need to maintain any format for those names ?
m
✅ thought for 342.1 seconds
Short answer: yes—when running a deployment by name, you must use the format <FLOW_NAME>/<DEPLOYMENT_NAME>. If your flow is named grammar-processor and your deployment is grammar-processor-deployment, the command should be:
Copy code
prefect deployment run "grammar-processor/grammar-processor-deployment" --watch
Common causes of “Failed to read deployment by name”: - Wrong format: You must include the slash: 
<flow>/<deployment>
. Passing just the deployment name won’t work. - Typos or name mismatch: The actual registered flow name may differ. By default, 
@flow
derives the name from the function name but replaces underscores with hyphens. For example, 
def grammar_processor():
becomes a flow named 
grammar-processor
. - Deployment doesn’t exist in your current workspace/profile: You might be pointed at a different API URL/workspace. - Extra whitespace/quotes issues: Quoting the name as shown above is safe. Quick checks: - List what Prefect sees (verify exact names):
Copy code
prefect deployment ls
- Inspect exactly the deployment you’re trying to run:
Copy code
prefect deployment inspect "grammar-processor/grammar-processor-deployment"
- Confirm your active profile and API/workspace:
Copy code
prefect profile inspect
prefect profile ls
How to create a matching deployment (Prefect 3.x):
Copy code
from prefect import flow

@flow
def grammar_processor():
    return "ok"

if __name__ == "__main__":
    grammar_processor.deploy(
        name="grammar-processor-deployment",
        work_pool_name="grammar-processor",
    )
Then run:
Copy code
prefect deployment run "grammar-processor/grammar-processor-deployment" --watch
If this still fails, please share: - The exact command you ran - The output of 
prefect deployment ls
- The output of 
prefect deployment inspect "grammar-processor/grammar-processor-deployment"
- The output of 
prefect profile inspect
I’ll pinpoint the issue from there.
14 Views
Open in Slack
PreviousNext
Powered by