Moving away from dbt core operation blocks

docker/Dockerfile.job-runner

@@ -0,0 +1,30 @@
+# Define version as build arg with default
+ARG PREFECT_VERSION=3.6.29
+
+FROM prefecthq/prefect:${PREFECT_VERSION}-python3.10-kubernetes
+
+# Shared volume mount point for client dbt project directories.
+ARG CLIENTDBT_ROOT="/mnt/appdata/clientdbts"
+ENV CLIENTDBT_ROOT=${CLIENTDBT_ROOT}
+
+# System deps — git is needed for the prefect-airbyte git dep pinned in pyproject.toml.
+RUN apt-get update && apt-get install -y \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+# Install all runtime deps from prefect-proxy's pyproject.toml + uv.lock. Single
+# source of truth — bumping dbt-core / prefect-dbt / adapters in pyproject flows
+# into the image on the next build with no Dockerfile edit. uv exports the
+# frozen lockfile to requirements.txt so pip installs the resolved-and-locked
+# set into the base image's system Python.
+#
+# Build context is prefect-proxy/ (the parent of docker/) so pyproject.toml
+# and uv.lock are reachable — see docker/README.md.
+COPY pyproject.toml uv.lock ./
+RUN pip install --no-cache-dir uv \
+    && uv export --no-dev --frozen --no-hashes --format requirements-txt --output-file /tmp/req.txt \
+    && pip install --no-cache-dir -r /tmp/req.txt \
+    && rm /tmp/req.txt pyproject.toml uv.lock
+
+# Client dbt projects get mounted here at runtime.
+RUN mkdir -p ${CLIENTDBT_ROOT} && mkdir -p /root/.dbt

docker/README.md

@@ -0,0 +1,85 @@
+# Prefect Job Runner Docker Image
+
+Docker image for running Dalgo Prefect flows in EKS.
+
+## What is this?
+
+This image runs Dalgo Prefect flows on EKS. It installs all runtime deps
+(prefect, prefect-dbt, prefect-shell, prefect-airbyte, dbt-core + adapters,
+elementary-data, etc.) directly from `prefect-proxy/pyproject.toml` + `uv.lock`
+— **single source of truth**. Bumping any dependency in pyproject flows into
+the image on the next build with no Dockerfile edit.
+
+`PrefectDbtRunner` uses `dbt-core` from the base env's Python — no per-version
+dbt venvs baked into the image.
+
+## How to Build
+
+### Build Arguments
+
+- `PREFECT_VERSION`: drives the base image tag (default: `3.6.29`). Must match
+  the `prefect==` pin in `pyproject.toml` — otherwise pip will upgrade prefect
+  on top of the base image.
+- `CLIENTDBT_ROOT`: mount path for client dbt projects (default:
+  `/mnt/appdata/clientdbts`).
+
+### Image tags
+
+Tags track the Prefect upgrade:
+
+- `0.1` → Prefect 3.1.15
+- `0.2` → Prefect 3.6.29
+
+### Environment-Specific Builds
+
+Dalgo's EKS runs on ARM nodes — build for `linux/arm64` for prod, `linux/amd64`
+for local x86 testing.
+
+**Build context is `prefect-proxy/`** (the parent of `docker/`) so
+`pyproject.toml` and `uv.lock` are reachable.
+
+```bash
+# From the prefect-proxy directory
+cd /path/to/prefect-proxy
+
+# Prod build (EKS, ARM)
+docker build --platform linux/arm64 -f docker/Dockerfile.job-runner \
+  --build-arg PREFECT_VERSION=3.6.29 \
+  -t tech4dev/prefect-eks-job-runner:0.2 .
+
+# Local x86 test build
+docker build --platform linux/amd64 -f docker/Dockerfile.job-runner \
+  --build-arg PREFECT_VERSION=3.6.29 \
+  -t tech4dev/prefect-eks-job-runner:0.2-amd64 .
+
+# Multi-platform build + push (single tag, both arches)
+docker buildx build --platform linux/arm64,linux/amd64 \
+  -f docker/Dockerfile.job-runner \
+  --build-arg PREFECT_VERSION=3.6.29 \
+  -t tech4dev/prefect-eks-job-runner:0.2 \
+  --push .
+
+# Build with a custom shared-volume mount point
+docker build --platform linux/arm64 -f docker/Dockerfile.job-runner \
+  --build-arg PREFECT_VERSION=3.6.29 \
+  --build-arg CLIENTDBT_ROOT=/dev/client/dbt \
+  -t tech4dev/prefect-eks-job-runner:<tag> .
+```
+
+Sanity check the version pins inside a built image:
+```bash
+docker run --rm --platform linux/arm64 tech4dev/prefect-eks-job-runner:0.2 python -c "
+import dbt.version, importlib.metadata as m
+print('dbt-core:     ', dbt.version.__version__)
+print('dbt-postgres: ', m.version('dbt-postgres'))
+print('dbt-bigquery: ', m.version('dbt-bigquery'))
+"
+```
+Values must match the pins in `prefect-proxy/pyproject.toml`. If they drift,
+the lockfile export or install step isn't taking.
+
+## Usage
+
+Set this image in your EKS work pool configuration. Existing deployment
+parameters work unchanged since the directory structure
+(`/mnt/appdata/clientdbts/`) is preserved.

proxy/helpers.py

@@ -101,6 +101,7 @@ def deployment_to_json(deployment: dict) -> dict:
         "cron": "",
         "isScheduleActive": False,
         "parameters": deployment["parameters"],
+        "entrypoint": deployment.get("entrypoint"),
     }
     if deployment.get("schedules") and len(deployment["schedules"]) > 0:
         retval["cron"] = deployment["schedules"][0]["schedule"]["cron"]

proxy/main.py

@@ -7,6 +7,7 @@
 from prefect_airbyte import AirbyteConnection
 import sentry_sdk
 from proxy.helpers import CustomLogger, deployment_to_json
+from proxy.exception import PrefectException
 
 
 from proxy.service import (
@@ -15,6 +16,7 @@
     create_airbyte_server_block,
     create_dbt_core_block,
     put_deployment_v1,
+    update_deployment_entrypoint,
     update_postgres_credentials,
     update_bigquery_credentials,
     update_target_configs_schema,
@@ -32,6 +34,7 @@
     get_flow_run,
     retry_flow_run,
     create_secret_block,
+    get_secret_block_by_name,
     upsert_secret_block,
     _create_dbt_cli_profile,
     update_dbt_cli_profile,
@@ -70,10 +73,7 @@
     FilterPrefectWorkers,
 )
 
-from proxy.prefect_flows import (
-    run_shell_operation_flow,
-    run_dbtcore_flow_v1,
-)
+from proxy.prefect_flows_runner import dbtjob_v2_runner, shellopjob
 
 
 sentry_sdk.init(
@@ -107,14 +107,11 @@ def dbtrun_v1(task_config: RunDbtCoreOperation):
         raise TypeError("invalid task config")
     logger.info("dbt core operation running %s", task_config.slug)
 
-    flow = run_dbtcore_flow_v1
-    if task_config.flow_name:
-        flow = flow.with_options(name=task_config.flow_name)
-    if task_config.flow_run_name:
-        flow = flow.with_options(flow_run_name=task_config.flow_run_name)
-
+    # Ignore payload.flow_name / payload.flow_run_name — use whatever the
+    # runner decorator declares (name="dbtjob_v2_runner",
+    # flow_run_name="dbtjob-{task_slug}").
     try:
-        result = flow(task_config.model_dump())
+        result = dbtjob_v2_runner(task_config.model_dump(), task_config.slug)
         return result
     except Exception as error:
         logger.exception(error)
@@ -128,14 +125,10 @@ def shelloprun(task_config: RunShellOperation):
     if not isinstance(task_config, RunShellOperation):
         raise TypeError("invalid task config")
 
-    flow = run_shell_operation_flow
-    if task_config.flow_name:
-        flow = flow.with_options(name=task_config.flow_name)
-    if task_config.flow_run_name:
-        flow = flow.with_options(flow_run_name=task_config.flow_run_name)
-
+    # Ignore payload.flow_name / payload.flow_run_name — use the runner
+    # decorator's (name="shellopjob", flow_run_name="shellop-{task_slug}").
     try:
-        result = flow(task_config.model_dump())
+        result = shellopjob(task_config.model_dump(), task_config.slug)
         return result
     except Exception as error:
         logger.exception(error)
@@ -369,6 +362,23 @@ async def put_dbtcore_schema(payload: DbtCoreSchemaUpdate):
     return {"success": 1}
 
 
+# =============================================================================
+@app.get("/proxy/blocks/secret/{blockname}")
+async def get_secret_block(blockname: str):
+    """Look up a Secret block by name. Returns {block_id, block_name} or 404
+    if not found."""
+    if not isinstance(blockname, str):
+        raise TypeError("blockname is invalid")
+    try:
+        result = await get_secret_block_by_name(blockname)
+    except PrefectException as error:
+        raise HTTPException(status_code=404, detail=str(error)) from error
+    except Exception as error:
+        logger.exception(error)
+        raise HTTPException(status_code=400, detail="failed to fetch secret block") from error
+    return result
+
+
 # =============================================================================
 @app.post("/proxy/blocks/secret/")
 async def post_secret_block(payload: PrefectSecretBlockCreate):
@@ -490,6 +500,22 @@ def put_dataflow_v1(deployment_id, payload: DeploymentUpdate2):
     return {"success": 1}
 
 
+@app.patch("/proxy/v1/deployments/{deployment_id}/entrypoint")
+def patch_deployment_entrypoint(deployment_id: str, payload: dict):
+    """PATCH a deployment's entrypoint. Used by the runner-flow migration script."""
+    entrypoint = payload.get("entrypoint") if isinstance(payload, dict) else None
+    if not isinstance(entrypoint, str) or not entrypoint:
+        raise HTTPException(status_code=400, detail="entrypoint (string) is required")
+    try:
+        update_deployment_entrypoint(deployment_id, entrypoint)
+    except Exception as error:
+        logger.exception(error)
+        raise HTTPException(
+            status_code=400, detail="failed to update deployment entrypoint"
+        ) from error
+    return {"success": 1}
+
+
 @app.get("/proxy/v1/deployments/get_scheduled_flow_runs")
 def get_dataflow_scheduled_flow_runs(deployment_id: str):
     """fetch scheduled flow-runs for a deployment"""

proxy/prefect_flows.py

@@ -4,6 +4,7 @@
 everything under here is incremented by a version compared to flows.py
 """
 
+import json
 import os
 from time import sleep
 import asyncio
@@ -314,12 +315,14 @@ def shellopjob(task_config: dict, task_slug: str):  # pylint: disable=unused-arg
     elif task_config["slug"] == "generate-edr":  # DDP_backend:constants.TASK_GENERATE_EDR
         # commands = ["edr send-report --bucket-file-path reports/{orgname}.TODAYS_DATE.html --profiles-dir elementary_profiles"]
         # env = {"PATH": /path/to/dbt/venv, "shell": "/bin/bash"}
-        secret_block_aws_access_key = "edr-aws-access-key"
-        aws_access_key = Secret.load(secret_block_aws_access_key).get()
-        secret_block_aws_access_secret = "edr-aws-access-secret"
-        aws_access_secret = Secret.load(secret_block_aws_access_secret).get()
-        secret_block_s3_bucket = "edr-s3-bucket"
-        edr_s3_bucket = Secret.load(secret_block_s3_bucket).get()
+        # Single consolidated Secret block replaces the old trio
+        # (edr-aws-access-key / edr-aws-access-secret / edr-s3-bucket).
+        # Backend scaffolds it via `manage.py sync_edr_secret_block`.
+        raw = Secret.load("edr-s3-creds").get()
+        edr_config = raw if isinstance(raw, dict) else json.loads(raw)
+        aws_access_key = edr_config["aws_access_key_id"]
+        aws_access_secret = edr_config["aws_secret_access_key"]
+        edr_s3_bucket = edr_config["s3_bucket"]
         # object key for the report
         todays_date = datetime.today().strftime("%Y-%m-%d")
         task_config["commands"][0] = task_config["commands"][0].replace("TODAYS_DATE", todays_date)