Thanks for the context — this is doable with Prefect 3.x, but there are two different patterns depending on whether you want to keep using a process work pool or pivot to a docker work pool. Since you’re on a process pool, here’s the recommended approach: Key points up front - Process workers do not pull a container per flow run. They execute the flow as a local subprocess on the worker machine. If you want a containerized environment, you run the worker inside your own Docker image. - You can still “activate” a virtualenv per flow by either: - Overriding the job’s command to source the venv then call Prefect, or - Simpler/safer: inject

VIRTUAL_ENV

and prepend the venv’s

bin

to

PATH

via

job_variables.env

and let the worker use the default command. Good docs to keep handy: - prefect.yaml - Customize job variables - Process worker API - Deployments concepts Option A (recommended with your process worker): build a “worker image” with all flows + venvs 1) Build a single worker image that: - Copies your flow code into the image - Pre-creates a virtualenv per flow and installs each flow’s requirements (including Prefect) - You run your process worker inside this image Example Dockerfile (worker)

# syntax=docker/dockerfile:1
ARG PYTHON_VERSION=3.11
FROM python:${PYTHON_VERSION}-slim

ENV PIP_NO_CACHE_DIR=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    PYTHONUNBUFFERED=1

# Install any system deps your flows need
RUN apt-get update && apt-get install -y --no-install-recommends \
    bash ca-certificates curl git \
 && rm -rf /var/lib/apt/lists/*

# Create directories for app and venvs
WORKDIR /app
RUN mkdir -p /opt/venvs

# Copy your repository (adjust as needed)
# Layout assumed:
#   flows/
#     myflow_a/
#       flow.py
#       requirements.txt
#     myflow_b/
#       flow.py
#       requirements.txt
COPY . /app

# Create per-flow virtualenvs and install requirements (including Prefect)
# Repeat for each flow dir; adjust names/paths to your repo structure
RUN python -m venv /opt/venvs/myflow_a \
 && /opt/venvs/myflow_a/bin/pip install --upgrade pip \
 && /opt/venvs/myflow_a/bin/pip install "prefect>=3" -r flows/myflow_a/requirements.txt

RUN python -m venv /opt/venvs/myflow_b \
 && /opt/venvs/myflow_b/bin/pip install --upgrade pip \
 && /opt/venvs/myflow_b/bin/pip install "prefect>=3" -r flows/myflow_b/requirements.txt

# Optional: create a non-root user
# RUN useradd -m -u 10001 prefect && chown -R prefect:prefect /app /opt/venvs
# USER prefect

# Default command is just a placeholder; you’ll pass the worker start command at runtime
CMD ["bash", "-lc", "prefect version"]

How to run the worker container - Create a process work pool (once):

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

- Start your worker using this image (e.g., in Docker or Kubernetes) and point it at your pool:

docker run --rm -it \
  -e PREFECT_API_URL=<your_api_url> \
  -e PREFECT_API_KEY=<your_api_key> \
  <http://your-registry.example.com/prefect/worker:latest|your-registry.example.com/prefect/worker:latest> \
  bash -lc "prefect worker start --pool my-process-pool --name my-worker"

prefect.yaml for deployments (process worker) - Use env injection to select the right venv per flow - Keep

command

empty so the worker generates the correct execution command ``` # prefect.yaml name: my-project # No build/push needed for process workers; the worker image runs your code # If you still want to build/push this image with Prefect steps, see Option B below. pull: # If your code is baked into the worker image at /app, you can just set the working dir - prefect.deployments.steps.pull.set_working_directory: directory: /app deployments: - name: myflow-a

entrypoint: flows/myflow_a/flow.py:flow work_pool: name: my-process-pool job_variables: working_dir: /app # Do NOT set 'command' so the worker uses the default engine command env: VIRTUAL_ENV: /opt/venvs/myflow_a PATH: /opt/venvs/myflow_a/bin:{{ env.PATH }} - name: myflow-b entrypoint: flows/myflow_b/flow.py:flow work_pool: name: my-process-pool job_variables: working_dir: /app env: VIRTUAL_ENV: /opt/venvs/myflow_b PATH: /opt/venvs/myflow_b/bin:{{ env.PATH }}

Notes
- The process worker’s default command is generated automatically when `job_variables.command` is unset. Injecting `VIRTUAL_ENV` and `PATH` causes the default command to use the venv’s Python/Prefect.
- If you truly need to “activate” the venv via bash, you can set:

job_variables: command: bash -lc 'source /opt/venvs/myflow_a/bin/activate && prefect flow-run execute'

The worker injects the flow run context for `prefect flow-run execute` so an explicit ID argument isn’t necessary.

Option B (if you do want per-flow containers pulled at runtime): use a docker work pool
- This aligns with “build image per flow, push to Artifactory, pull per run”.
- Use `prefect_docker.deployments.steps` in `prefect.yaml` to build and push, and set the Docker image on your deployment.

Example prefect.yaml (docker worker)

name: my-project build: - prefect_docker.deployments.steps.build_docker_image: id: build_image image_name: artifactory.example.com/prefect/myflow-a dockerfile: Dockerfile tag: "{{ env.GIT_SHA | default('latest') }}" push: - prefect_docker.deployments.steps.push_docker_image: image_name: artifactory.example.com/prefect/myflow-a tag: "{{ build_image.tag }}" # If your Docker host is already logged in to Artifactory, you can omit credentials. # Otherwise, ensure auth via docker login or provide credentials here if your environment requires it. pull: # At runtime, code is in the container; optionally set working dir - prefect.deployments.steps.pull.set_working_directory: directory: /app deployments: - name: myflow-a entrypoint: flows/myflow_a/flow.py:flow work_pool: name: my-docker-pool job_variables: image: "{{ build_image.image }}" env: VIRTUAL_ENV: /opt/venvs/myflow_a PATH: /opt/venvs/myflow_a/bin:{{ env.PATH }}

And the corresponding Dockerfile (per-flow)

FROM python:3.11-slim WORKDIR /app COPY flows/myflow_a /app/flows/myflow_a RUN python -m venv /opt/venvs/myflow_a \ && /opt/venvs/myflow_a/bin/pip install --upgrade pip \ && /opt/venvs/myflow_a/bin/pip install "prefect>=3" -r /app/flows/myflow_a/requirements.txt ENV VIRTUAL_ENV=/opt/venvs/myflow_a ENV PATH=/opt/venvs/myflow_a/bin:$PATH # No need for a custom entrypoint; the Docker worker will run your flow ``` Which option should you choose? - Stay with Option A (process pool) if you want simple local subprocess execution and are okay maintaining a long-running worker container image that contains all your code/venvs. - Choose Option B (docker pool) if you want each flow run fully containerized and pulled from Artifactory at runtime. Questions for you - Do you want to stick with the process worker model, or would you prefer per-run containers (docker work pool)? - How are your flows laid out on disk (so we can tailor the COPY and venv creation in the Dockerfile)? - Any org constraints for Artifactory auth (e.g., you need the build to pass explicit credentials vs relying on docker login)? If you share the repo layout and flow names, I can tailor the Dockerfile and prefect.yaml exactly to your structure.

✅ thought for 445.3 seconds

Great — moving to a Kubernetes work pool with a private image is a solid choice. Below is a refactor of your prefect.yaml for Prefect 3.x, a Dockerfile that uses a virtual environment, plus the Kubernetes bits you’ll need for imagePullSecrets. Key changes - Containerize dependencies: we remove runtime

pip_install_requirements

and install deps in the image. - Keep Git-based code updates: we retain a

pull

git_clone step so you don’t need to rebuild the image for code-only changes. - Kubernetes pool: set

job_variables.image

to your private image and configure the pool’s base job template with

imagePullSecrets

. Refactored prefect.yaml (Prefect 3.x)

name: common
prefect-version: 3.4.21

# Build/push the image that your K8s jobs will run
# Requires prefect-docker on the machine running `prefect deploy`
build:
  - prefect_docker.deployments.steps.build_docker_image:
      id: build-image
      image_name: <your-registry>/minerva-prefect
      tag: latest
      dockerfile: ./Dockerfile
      context: .
push:
  - prefect_docker.deployments.steps.push_docker_image:
      # Ensures the push step is available; install on your deploy machine
      requires: "prefect-docker>=0.3.11"
      image: "{{ build-image.image }}"

# Pull your code at runtime so you can deploy code updates without rebuilding the image
pull:
  - prefect.deployments.steps.git_clone:
      id: clone-step
      repository: <https://github.com/OneAdobe/minerva-prefect.git>
      branch: main
      access_token: "{{ prefect.blocks.secret.dxminerva-git-pat }}"

deployments:
  - name: "enrich_ims_org_info_table"
    entrypoint: "flows/common/enrich_ims_org_info_table/flow.py:enrich_ims_org_info_table"
    work_pool:
      name: "k8s-pool"           # <-- change to your Kubernetes work pool name
      job_variables:
        image: "{{ build-image.image }}"
        imagePullPolicy: IfNotPresent
    schedules:
      - cron: "0 11 * * *"
        slug: "enrich_ims_org_info_table-daily-at-11am-ist"
        timezone: "Asia/Kolkata"
        active: true

  - name: "get_export_job_stats"
    entrypoint: "flows/ups_profile_export/get_export_job_stats/flow.py:get_export_job_stats"
    work_pool:
      name: "k8s-pool"
      job_variables:
        image: "{{ build-image.image }}"
        imagePullPolicy: IfNotPresent
    schedules:
      - cron: "30 11 * * *"
        slug: "get_export_job_stats-daily-at-1130am-ist"
        timezone: "Asia/Kolkata"
        active: true
        parameters:
          envs:
            - prod
          regions:
            - va7
            - nld2
            - aus5
            - can2
            - gbr9
            - ind2
          models:
            - "_xdm.context.profile"
          sources:
            - "scheduler"

  - name: "sandbox_capacity_anomaly_detection"
    entrypoint: "flows/axia/sandbox_capacity_anomaly_detection/flow.py:sandbox_capacity_anomaly_report"
    work_pool:
      name: "k8s-pool"
      job_variables:
        image: "{{ build-image.image }}"
        imagePullPolicy: IfNotPresent
    schedules:
      - cron: "00 12 * * *"
        slug: "sandbox_capacity_anomaly_detection-daily-at-1200pm-ist"
        timezone: "Asia/Kolkata"
        active: true
        parameters:
          envs:
            - prod
          regions:
            - va7
            - nld2
            - aus5
            - can2
            - gbr9
            - ind2
          capacity_names:
            - STREAMING_THROUGHPUT
          slack_channel: C09LZBQ1K1V

Dockerfile (with venv) ``` # Python base FROM python:3.11-slim # System prep ENV PYTHONDONTWRITEBYTECODE=1 ENV PYTHONUNBUFFERED=1 # Create virtual environment ENV VENV=/opt/venv RUN python -m venv $VENV && $VENV/bin/pip install --upgrade pip wheel setuptools ENV PATH="$VENV/bin:$PATH" # Install system deps (git for runtime git_clone step) RUN apt-get update && apt-get install -y --no-install-recommends git \ && rm -rf /var/lib/apt/lists/*

# Install Prefect runtime and your flow dependencies # - Keep Prefect aligned with your prefect-version # - Use a consolidated requirements.txt at repo root (or adjust as needed) COPY requirements.txt /tmp/requirements.txt RUN pip install --no-cache-dir "prefect==3.4.21" \ && pip install --no-cache-dir -r /tmp/requirements.txt # Optional: set a working directory for any runtime files WORKDIR /opt/prefect

Notes on requirements
- Because we removed runtime `pip_install_requirements`, put all Python dependencies needed by these three flows into a single repo-level requirements.txt that you COPY into the image.
- If you truly need different deps per flow, we can split into per-deployment images (one build/push per deployment) — happy to provide that variant.

Kubernetes private registry: imagePullSecrets
1) Create the Kubernetes secret (replace values and namespace):

kubectl create secret docker-registry my-registry-secret \ --docker-server=<your-registry> \ --docker-username=<username> \ --docker-password=<password or token> \ --docker-email=<email@example.com> \ -n <your-prefect-namespace>

2) Configure the work pool’s base job template to use that secret. Create base-job-template.json:

{
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "spec": {
        "imagePullSecrets": [
          {
            "name": "my-registry-secret"
          }
        ]
      }
    }
  }
}

3) Create or update your Kubernetes work pool to include the template, then start a worker:

# Create the pool (once) prefect work-pool create "k8s-pool" --type kubernetes # Update base job template to add imagePullSecrets prefect work-pool update "k8s-pool" --base-job-template ./base-job-template.json # Start a worker (ensure the host has Kubernetes access) # If you run this in an environment without the extras preinstalled: uv run --with 'prefect[kubernetes]' prefect worker start --pool "k8s-pool"

Deploying
- From your project root (where prefect.yaml and Dockerfile live), run:

prefect deploy --no-prompt ``` - This will build and push your image once, then create/update the three deployments. - The

pull

git_clone step will fetch your repo at runtime; since deps come from the image, no pip install is done during execution. Placeholders to update -

<your-registry>

in prefect.yaml and the Dockerfile build-image.image_name. - Kubernetes secret name and namespace in base-job-template.json and kubectl command. - The work pool name if you choose a different one than "k8s-pool". Docs and references - Prefect YAML and deployments: prefect.yaml format and create deployments - Schedules: scheduling - Kubernetes work pools: Kubernetes deployment guide and manage work pools - CLI deploy: prefect deploy CLI Want me to switch this to per-deployment images (so each flow can keep its own requirements.txt) or is a single shared image okay for your case?

Short answer - Pass env vars to the Kubernetes Job via deployment-level job_variables or via the Kubernetes work pool’s base job template. Helm chart values affect the worker container itself; they don’t automatically propagate to the Job pods the worker spawns. You can, however, use Helm to run a post-install/upgrade command that updates the work pool’s base job template so all jobs inherit your env. Three ways to set env for K8s jobs 1) Per-deployment (prefect.yaml) via job_variables - Good for flow/deployment-specific configuration.

deployments:
  - name: my-deployment
    entrypoint: flows/my_flow.py:my_flow
    work_pool:
      name: k8s-pool
      job_variables:
        env:
          - name: APP_ENV
            value: prod
          - name: SECRET_TOKEN
            valueFrom:
              secretKeyRef:
                name: my-secret
                key: token
        # If you prefer bulk injection:
        envFrom:
          - secretRef:
              name: shared-secrets
          - configMapRef:
              name: shared-config

2) Work pool defaults via base job template - Best for org-wide defaults that every job should inherit (e.g., shared env, envFrom, volumes, annotations, serviceAccountName). - Put env under the container spec in the base job template and update the pool.

# base-job-template.json (minimal example)
{
  "apiVersion": "batch/v1",
  "kind": "Job",
  "spec": {
    "template": {
      "metadata": {
        "labels": { "app": "prefect-flow" }
      },
      "spec": {
        "serviceAccountName": "prefect-worker",
        "imagePullSecrets": [{ "name": "my-registry-secret" }],
        "containers": [
          {
            "name": "flow",
            "env": [
              { "name": "APP_ENV", "value": "prod" },
              {
                "name": "SECRET_TOKEN",
                "valueFrom": { "secretKeyRef": { "name": "my-secret", "key": "token" } }
              }
            ],
            "envFrom": [
              { "secretRef": { "name": "shared-secrets" } },
              { "configMapRef": { "name": "shared-config" } }
            ]
          }
        ]
      }
    }
  }
}

Then apply it to the pool:

prefect work-pool update "k8s-pool" --base-job-template ./base-job-template.json

Precedence: per-run overrides > deployment job_variables > work pool base job template. 3) Via Helm (indirect) - Helm values directly set env on the Worker Deployment pod, not on the Jobs it spawns. - To have Helm “apply defaults” to the Jobs, add a Helm hook or post-install/upgrade job that runs:

prefect work-pool update "k8s-pool" --base-job-template /mounted/base-job-template.json

- Store your base-job-template.json in a ConfigMap and mount it into that hook job. This way, the pool’s template (not the worker’s env) defines env/envFrom for all jobs. Notes - Private images: keep imagePullSecrets in the base job template so every job can pull your private image. - You can still add deployment-specific env via job_variables even when a base template exists; they merge, with deployment values taking precedence. Docs - Kubernetes work pools: Kubernetes deployment guide - Manage work pools: work pool configuration - Customize job variables: job variables - Prefect Helm chart repo: prefect-helm If you share your Helm chart/values structure, I can sketch a ConfigMap + post-upgrade Job snippet to keep the pool’s base job template in sync automatically.

You’re right: with your current prefect.yaml, the build/push steps are being executed for each deployment. That’s because Prefect evaluates step outputs in the context of each deployment when they’re referenced (e.g.,

{{ build-image.image }}

inside

job_variables

), so the build pipeline is rerun. Two solid ways to build the image only once Option A — Deploy all in one invocation (still using build/push steps) - Run a single

prefect deploy

for all deployments in the file (no filters/entrypoints). The build/push steps will run once per invocation. - If your CI currently calls

prefect deploy

three times (once per flow), change it to a single call:

prefect deploy --no-prompt

If you must target a subset, you can still do it in one invocation with a pattern for

--name

, but the safest/clearest is “no args = all deployments”. Option B — Move image build/push to CI, remove build/push from prefect.yaml - This is the most reliable way to guarantee one build. Build and push the image in CI (using your commit SHA tag), then pass the image reference into deployments via env vars. Prefect won’t rebuild anything. Example CI snippet

export PREFECT_IMAGE_NAME=<your-registry>/minerva-prefect
export PREFECT_IMAGE_TAG=$(git rev-parse --short HEAD)

docker build -t "${PREFECT_IMAGE_NAME}:${PREFECT_IMAGE_TAG}" -f Dockerfile .
docker push "${PREFECT_IMAGE_NAME}:${PREFECT_IMAGE_TAG}"

# Create or update all deployments in one shot, using the env vars
prefect deploy --no-prompt

Updated prefect.yaml for Option B (no build/push; use a shared image)

name: minerva-prefect
prefect-version: 3.4.21

# No build/push sections here; CI handles it

definitions:
  tags: &common_tags
    - "eks"
    - "{{ $PREFECT_IMAGE_TAG }}"    # reuse the tag in your deployment tags if useful
  work_pool: &common_work_pool
    name: "minerva-kubernetes"
    job_variables:
      image: "{{ $PREFECT_IMAGE_NAME }}:{{ $PREFECT_IMAGE_TAG }}"
      imagePullPolicy: IfNotPresent

deployments:
  - name: "enrich_ims_org_info_table"
    entrypoint: "flows/common/enrich_ims_org_info_table/flow.py:enrich_ims_org_info_table"
    tags: *common_tags
    work_pool: *common_work_pool
    schedules:
      - cron: "0 11 * * *"
        slug: "enrich_ims_org_info_table-daily-at-11am-ist"
        timezone: "Asia/Kolkata"
        active: true

  - name: "get_export_job_stats"
    entrypoint: "flows/ups_profile_export/get_export_job_stats/flow.py:get_export_job_stats"
    tags: *common_tags
    work_pool: *common_work_pool
    schedules:
      - cron: "30 11 * * *"
        slug: "get_export_job_stats-daily-at-1130am-ist"
        timezone: "Asia/Kolkata"
        active: true
        parameters:
          envs: [prod]
          regions: [va7, nld2, aus5, can2, gbr9, ind2]
          models: ["_xdm.context.profile"]
          sources: ["scheduler"]

  - name: "sandbox_capacity_anomaly_detection"
    entrypoint: "flows/axia/sandbox_capacity_anomaly_detection/flow.py:sandbox_capacity_anomaly_report"
    tags: *common_tags
    work_pool: *common_work_pool
    schedules:
      - cron: "00 12 * * *"
        slug: "sandbox_capacity_anomaly_detection-daily-at-1200pm-ist"
        timezone: "Asia/Kolkata"
        active: true
        parameters:
          envs: [prod]
          regions: [va7, nld2, aus5, can2, gbr9, ind2]
          capacity_names: [STREAMING_THROUGHPUT]
          slack_channel: C09LZBQ1K1V

Why this works - Prefect won’t run any build/push steps; it will just create/update deployments that point to the already-pushed image. - You still get a single image for all deployments (tagged by commit), and

prefect deploy

becomes fast and idempotent. If you prefer to keep build/push inside Prefect and still avoid multiple builds, confirm that your CI is running a single

prefect deploy

invocation for all deployments. If you’re already doing that and still seeing multiple builds, I recommend the CI-based build approach above. Refs

- Prefect YAML guide: prefect.yaml format - CLI deploy reference: prefect deploy CLI If you share how you’re currently calling

prefect deploy

in CI, I can suggest the smallest change to make it build once.