Moving away from dbt core operation blocks - #231
Conversation
WalkthroughThe PR adds a Prefect job-runner image, migrates execution to runner-based flows, adds dbt profile and orchestration logic, exposes secret and deployment entrypoint APIs, consolidates EDR credentials, and updates dependencies and tests. ChangesRuntime image and dependencies
Runner migration and orchestration
API and deployment integration
EDR credentials
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant APIClient
participant proxy_main
participant deployment_schedule_flow_v5
participant TaskRunner
participant PrefectDbtRunner
APIClient->>proxy_main: submit dbt or shell task
proxy_main->>TaskRunner: call runner with payload and task slug
TaskRunner->>PrefectDbtRunner: invoke dbt command
PrefectDbtRunner-->>TaskRunner: return result or failure state
TaskRunner-->>proxy_main: return execution outcome
proxy_main-->>APIClient: return response or HTTP 400
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #231 +/- ##
==========================================
- Coverage 70.58% 65.35% -5.23%
==========================================
Files 4 5 +1
Lines 1448 1723 +275
==========================================
+ Hits 1022 1126 +104
- Misses 426 597 +171 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
docker/Dockerfile.job-runner (1)
4-31: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winContainer runs as root — no
USERdirective.The image never drops privileges;
/root/.dbtand${CLIENTDBT_ROOT}are created and presumably written to as root at runtime. This is flagged by Trivy (DS-0002) and is a real security-posture gap for an EKS-scheduled workload that mounts client dbt projects and pulls client git repos.🔒 Suggested fix (adjust for HOME/volume ownership)
RUN mkdir -p ${CLIENTDBT_ROOT} && mkdir -p /root/.dbt + +RUN groupadd -r prefect && useradd -r -g prefect -d /home/prefect prefect \ + && mkdir -p /home/prefect/.dbt \ + && chown -R prefect:prefect ${CLIENTDBT_ROOT} /home/prefect +ENV HOME=/home/prefect +USER prefectNote: switching users requires the mounted
CLIENTDBT_ROOTvolume (and its subpaths written by git-clone/dbt tasks) to be writable by the new UID in the EKS work-pool volume config.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@docker/Dockerfile.job-runner` around lines 4 - 31, Update the Dockerfile to create and use a non-root runtime user, add a USER directive after dependency installation, and set HOME/dbt directory ownership and permissions for that user. Ensure CLIENTDBT_ROOT is writable by the selected UID, coordinating the corresponding EKS work-pool volume configuration so mounted client dbt projects and git-clone/dbt task subpaths remain writable.Source: Linters/SAST tools
🧹 Nitpick comments (4)
proxy/prefect_flows.py (1)
321-325: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd coverage for the consolidated EDR secret contract.
Test both dict and JSON-string Secret values, plus the three required keys. This changed report-delivery path is currently uncovered per the PR coverage report.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@proxy/prefect_flows.py` around lines 321 - 325, Add tests covering the EDR secret-loading logic around Secret.load("edr-s3-creds"): verify both dictionary and JSON-string secret values are accepted, and assert aws_access_key_id, aws_secret_access_key, and s3_bucket are read from the resulting configuration. Ensure the changed report-delivery path is exercised.proxy/prefect_flows_runner.py (1)
308-324: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winFragile exact-match parsing of macro stdout.
_extract_elementary_profile_from_macro_outputgathers lines only after an exactline == "elementary:"match. Any leading/trailing whitespace or minor formatting change in dbt's stdout forelementary.generate_elementary_cli_profile(e.g. across dbt-core/elementary version bumps) would silently produce an empty buffer, raising a somewhat genericRuntimeError. Consider a more tolerant match.♻️ Suggested tolerant match
- if line == "elementary:": + if line.strip() == "elementary:": gather = True🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@proxy/prefect_flows_runner.py` around lines 308 - 324, Update _extract_elementary_profile_from_macro_output to detect the elementary: YAML marker after trimming surrounding whitespace from each stdout line, while preserving the existing behavior of buffering from the first marker onward and raising when no marker is found.tests/test_prefect_flows_runner.py (2)
1-256: 📐 Maintainability & Code Quality | 🔵 Trivial | 🏗️ Heavy liftNo coverage for the dispatcher,
_prepare_elementary_profile, or generate-edr paths.
proxy/prefect_flows_runner.pyis reported at 38.02% patch coverage with 163 missing lines. This file covers only the pure helpers anddbtjob_v2_runner;_run_task_runner,_run_tasks_sequentially/_run_tasks_with_sync_tolerance,deployment_schedule_flow_v5,_prepare_elementary_profile, and thegenerate-edrbranch ofshellopjob— all new, critical execution paths — have no tests here. Worth prioritizing tests for the dispatcher (task-type routing, fail-fast vs sync-tolerant behavior) and the elementary profile builder (mockingsubprocess.runand macro stdout parsing).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_prefect_flows_runner.py` around lines 1 - 256, Expand tests beyond the existing dbtjob_v2_runner coverage to exercise _run_task_runner routing, sequential versus sync-tolerant task execution, deployment_schedule_flow_v5, _prepare_elementary_profile with mocked subprocess.run and macro-output parsing, and shellopjob’s generate-edr branch. Verify task-type dispatch, fail-fast and tolerance behavior, profile generation, and EDR execution outcomes using the existing mocking patterns.
224-241: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winAdd a multi-command dbt-test case to lock in the failure-classification contract.
This test only covers the single-command scenario. Given the misclassification risk flagged in
dbtjob_v2_runner(a non-test command failing under adbt-testslug being reported asDBT_TEST_FAILED), add a case withcommands=["dbt deps", "dbt test"]where the first command raises, asserting the exception propagates rather than being swallowed.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_prefect_flows_runner.py` around lines 224 - 241, Add a test alongside test_dbtjob_v2_runner_dbt_test_failure_returns_completed_state covering task configuration with commands=["dbt deps", "dbt test"], configuring the first command to raise; assert that exception propagates and is not converted to a DBT_TEST_FAILED result. Reuse the existing mocks and task helpers, preserving the single-command test’s behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@proxy/main.py`:
- Around line 506-510: Update the entrypoint validation in the deployment update
flow before update_deployment_entrypoint so whitespace-only strings are rejected
alongside missing and non-string values; validate the string’s trimmed content
while preserving the original entrypoint value for valid updates.
In `@proxy/prefect_flows_runner.py`:
- Around line 394-420: The external process calls in _prepare_elementary_profile
and shellopjob can block indefinitely. In proxy/prefect_flows_runner.py lines
394-420, add appropriate timeouts to both dbt deps and
elementary.generate_elementary_cli_profile subprocess.run calls and handle
subprocess.TimeoutExpired; also add a timeout and matching timeout handling to
the edr subprocess.run call in shellopjob’s generate-edr branch at lines
514-527.
- Around line 284-297: Restrict the DBT_TEST_FAILED tolerance in the command
loop to exceptions from the actual dbt test invocation, not merely tasks whose
slug is "dbt-test". Use the failing argv parsed for the current command to
identify the test command, while continuing to re-raise failures from
prerequisite commands such as deps or seed.
- Around line 568-605: Update the DBTCLOUD branch of _run_task_runner to avoid
calling asyncio.run for dbtcloudjob_v1 when invoked from Prefect-managed flows.
Use the existing Prefect-compatible async execution mechanism, or refactor the
call chain so the coroutine is awaited within deployment_schedule_flow_v5 while
preserving the current task dispatch behavior.
In `@proxy/service.py`:
- Around line 678-687: Update the GitRepository configuration in the deployment
creation flow around flow.from_source to use the production main branch instead
of the temporary feature/prefect-secret-blk branch. Ensure newly created
deployments pull the merged source; use an intentionally pinned release revision
only if that is the project’s established deployment convention.
---
Outside diff comments:
In `@docker/Dockerfile.job-runner`:
- Around line 4-31: Update the Dockerfile to create and use a non-root runtime
user, add a USER directive after dependency installation, and set HOME/dbt
directory ownership and permissions for that user. Ensure CLIENTDBT_ROOT is
writable by the selected UID, coordinating the corresponding EKS work-pool
volume configuration so mounted client dbt projects and git-clone/dbt task
subpaths remain writable.
---
Nitpick comments:
In `@proxy/prefect_flows_runner.py`:
- Around line 308-324: Update _extract_elementary_profile_from_macro_output to
detect the elementary: YAML marker after trimming surrounding whitespace from
each stdout line, while preserving the existing behavior of buffering from the
first marker onward and raising when no marker is found.
In `@proxy/prefect_flows.py`:
- Around line 321-325: Add tests covering the EDR secret-loading logic around
Secret.load("edr-s3-creds"): verify both dictionary and JSON-string secret
values are accepted, and assert aws_access_key_id, aws_secret_access_key, and
s3_bucket are read from the resulting configuration. Ensure the changed
report-delivery path is exercised.
In `@tests/test_prefect_flows_runner.py`:
- Around line 1-256: Expand tests beyond the existing dbtjob_v2_runner coverage
to exercise _run_task_runner routing, sequential versus sync-tolerant task
execution, deployment_schedule_flow_v5, _prepare_elementary_profile with mocked
subprocess.run and macro-output parsing, and shellopjob’s generate-edr branch.
Verify task-type dispatch, fail-fast and tolerance behavior, profile generation,
and EDR execution outcomes using the existing mocking patterns.
- Around line 224-241: Add a test alongside
test_dbtjob_v2_runner_dbt_test_failure_returns_completed_state covering task
configuration with commands=["dbt deps", "dbt test"], configuring the first
command to raise; assert that exception propagates and is not converted to a
DBT_TEST_FAILED result. Reuse the existing mocks and task helpers, preserving
the single-command test’s behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 55b5c386-137d-4f70-a6e2-590c8cc84b4b
⛔ Files ignored due to path filters (1)
uv.lockis excluded by!**/*.lock
📒 Files selected for processing (11)
docker/Dockerfile.job-runnerdocker/README.mdproxy/helpers.pyproxy/main.pyproxy/prefect_flows.pyproxy/prefect_flows_runner.pyproxy/service.pypyproject.tomlrequirements_dbt.txttests/test_main.pytests/test_prefect_flows_runner.py
💤 Files with no reviewable changes (1)
- requirements_dbt.txt
| 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) |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Reject whitespace-only entrypoints.
A value such as " " passes validation and overwrites a deployment with an unusable entrypoint.
Proposed fix
entrypoint = payload.get("entrypoint") if isinstance(payload, dict) else None
-if not isinstance(entrypoint, str) or not entrypoint:
+if not isinstance(entrypoint, str) or not entrypoint.strip():
raise HTTPException(status_code=400, detail="entrypoint (string) is required")
+entrypoint = entrypoint.strip()📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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) | |
| entrypoint = payload.get("entrypoint") if isinstance(payload, dict) else None | |
| if not isinstance(entrypoint, str) or not entrypoint.strip(): | |
| raise HTTPException(status_code=400, detail="entrypoint (string) is required") | |
| entrypoint = entrypoint.strip() | |
| try: | |
| update_deployment_entrypoint(deployment_id, entrypoint) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@proxy/main.py` around lines 506 - 510, Update the entrypoint validation in
the deployment update flow before update_deployment_entrypoint so
whitespace-only strings are rejected alongside missing and non-string values;
validate the string’s trimmed content while preserving the original entrypoint
value for valid updates.
| result = None | ||
| for cmd in task_config["commands"]: | ||
| argv = shlex.split(cmd)[1:] | ||
| try: | ||
| result = runner.invoke(argv) | ||
| except Exception: # pylint: disable=broad-exception-caught | ||
| if task_config["slug"] == "dbt-test": | ||
| return State( | ||
| type=StateType.COMPLETED, | ||
| name="DBT_TEST_FAILED", | ||
| message="WARNING: dbt test failed", | ||
| ) | ||
| raise | ||
| return result |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
dbt-test tolerance is keyed off the task slug, not the failing command.
The except fires for any command in task_config["commands"] that raises, and the check task_config["slug"] == "dbt-test" doesn't verify that the failing argv was actually a test invocation. If a dbt-test task's commands list ever includes a prior step (e.g. dbt deps/dbt seed) that fails, this would be silently reported as DBT_TEST_FAILED (a tolerated "warning") instead of propagating the real failure — masking data-integrity problems as test warnings. The current tests only cover the single-command case, so this gap is untested.
🐛 Suggested fix — scope tolerance to the actual failing command
- except Exception: # pylint: disable=broad-exception-caught
- if task_config["slug"] == "dbt-test":
+ except Exception: # pylint: disable=broad-exception-caught
+ if task_config["slug"] == "dbt-test" and argv and argv[0] == "test":
return State(
type=StateType.COMPLETED,
name="DBT_TEST_FAILED",
message="WARNING: dbt test failed",
)
raise📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| result = None | |
| for cmd in task_config["commands"]: | |
| argv = shlex.split(cmd)[1:] | |
| try: | |
| result = runner.invoke(argv) | |
| except Exception: # pylint: disable=broad-exception-caught | |
| if task_config["slug"] == "dbt-test": | |
| return State( | |
| type=StateType.COMPLETED, | |
| name="DBT_TEST_FAILED", | |
| message="WARNING: dbt test failed", | |
| ) | |
| raise | |
| return result | |
| result = None | |
| for cmd in task_config["commands"]: | |
| argv = shlex.split(cmd)[1:] | |
| try: | |
| result = runner.invoke(argv) | |
| except Exception: # pylint: disable=broad-exception-caught | |
| if task_config["slug"] == "dbt-test" and argv and argv[0] == "test": | |
| return State( | |
| type=StateType.COMPLETED, | |
| name="DBT_TEST_FAILED", | |
| message="WARNING: dbt test failed", | |
| ) | |
| raise | |
| return result |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@proxy/prefect_flows_runner.py` around lines 284 - 297, Restrict the
DBT_TEST_FAILED tolerance in the command loop to exceptions from the actual dbt
test invocation, not merely tasks whose slug is "dbt-test". Use the failing argv
parsed for the current command to identify the test command, while continuing to
re-raise failures from prerequisite commands such as deps or seed.
| # 4. dbt deps — installs elementary dbt package into dbt_packages/. | ||
| # Idempotent; safe if already installed. Fails loudly if packages.yml | ||
| # doesn't declare elementary (surface: user hasn't set up elementary yet). | ||
| logger.info("running dbt deps") | ||
| subprocess.run( | ||
| [dbt_bin, "deps", "--profiles-dir", "profiles"], | ||
| cwd=str(project_dir), | ||
| capture_output=True, | ||
| text=True, | ||
| check=True, | ||
| ) | ||
|
|
||
| # 5. Run the macro. | ||
| logger.info("running elementary.generate_elementary_cli_profile macro") | ||
| result = subprocess.run( | ||
| [ | ||
| dbt_bin, | ||
| "run-operation", | ||
| "elementary.generate_elementary_cli_profile", | ||
| "--profiles-dir", | ||
| "profiles", | ||
| ], | ||
| cwd=str(project_dir), | ||
| capture_output=True, | ||
| text=True, | ||
| check=True, | ||
| ) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unbounded subprocess.run calls — no timeouts anywhere in the elementary/edr path. All three external process invocations block their worker thread indefinitely if the child process hangs (e.g. network stall fetching the elementary dbt package, or edr hanging on S3 upload).
proxy/prefect_flows_runner.py#L394-L420: addtimeout=to both thedbt depsandelementary.generate_elementary_cli_profilesubprocess.runcalls in_prepare_elementary_profile, handlingsubprocess.TimeoutExpired.proxy/prefect_flows_runner.py#L514-L527: addtimeout=to theedrsubprocess.runcall inshellopjob's generate-edr branch.
🧰 Tools
🪛 ast-grep (0.45.0)
[error] 397-403: Command coming from incoming request
Context: subprocess.run(
[dbt_bin, "deps", "--profiles-dir", "profiles"],
cwd=str(project_dir),
capture_output=True,
text=True,
check=True,
)
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(subprocess-from-request)
[error] 407-419: Command coming from incoming request
Context: subprocess.run(
[
dbt_bin,
"run-operation",
"elementary.generate_elementary_cli_profile",
"--profiles-dir",
"profiles",
],
cwd=str(project_dir),
capture_output=True,
text=True,
check=True,
)
Note: [CWE-78] Improper Neutralization of Special Elements used in an OS Command ('OS Command Injection').
(subprocess-from-request)
🪛 Ruff (0.16.0)
[error] 398-398: subprocess call: check for execution of untrusted input
(S603)
[error] 408-408: subprocess call: check for execution of untrusted input
(S603)
📍 Affects 1 file
proxy/prefect_flows_runner.py#L394-L420(this comment)proxy/prefect_flows_runner.py#L514-L527
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@proxy/prefect_flows_runner.py` around lines 394 - 420, The external process
calls in _prepare_elementary_profile and shellopjob can block indefinitely. In
proxy/prefect_flows_runner.py lines 394-420, add appropriate timeouts to both
dbt deps and elementary.generate_elementary_cli_profile subprocess.run calls and
handle subprocess.TimeoutExpired; also add a timeout and matching timeout
handling to the edr subprocess.run call in shellopjob’s generate-edr branch at
lines 514-527.
| def _is_airbyte_sync_task(task_config: dict) -> bool: | ||
| """Check if a task is an airbyte sync task""" | ||
| return task_config["type"] == AIRBYTECONNECTION and task_config["slug"] == "airbyte-sync" | ||
|
|
||
|
|
||
| def _run_task_runner(task_config: dict): | ||
| """Copy of prefect_flows._run_task with DBTCORE and AIRBYTECONNECTION | ||
| branches dispatching to local runner-file versions; DBTCLOUD and | ||
| SHELLOPERATION still delegate to prefect_flows. | ||
| """ | ||
| if task_config["type"] == DBTCORE: | ||
| dbtjob_v2_runner(task_config, task_config["slug"]) | ||
|
|
||
| elif task_config["type"] == DBTCLOUD: | ||
| asyncio.run(dbtcloudjob_v1(task_config, task_config["slug"])) | ||
|
|
||
| elif task_config["type"] == SHELLOPERATION: | ||
| shellopjob(task_config, task_config["slug"]) | ||
|
|
||
| elif task_config["type"] == AIRBYTECONNECTION: | ||
| if task_config["slug"] == "airbyte-sync": | ||
| run_airbyte_connection_flow_v1(task_config) | ||
|
|
||
| elif task_config["slug"] == "airbyte-clear": | ||
| run_airbyte_conn_clear(task_config) | ||
|
|
||
| elif task_config["slug"] == "update-schema": | ||
| asyncio.run( | ||
| run_refresh_schema_flow( | ||
| task_config, catalog_diff=task_config.get("catalog_diff", {}) | ||
| ) | ||
| ) | ||
| else: | ||
| raise ValueError(f"Unsupported AIRBYTECONNECTION slug: {task_config['slug']}") | ||
|
|
||
| else: | ||
| raise ValueError(f"Unknown task type: {task_config['type']}") | ||
|
|
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
Does Prefect 3.6 run sync flows inside an already-running asyncio event loop?
💡 Result:
In Prefect 3.6, synchronous flows and tasks are designed to be executed without blocking or being directly wrapped inside the main asyncio event loop's execution context. Prefect employs internal utilities—specifically leveraging the greenback library—that allow synchronous code to execute even when called from an asynchronous context without blocking the event loop [1][2]. When a synchronous flow or task is executed, Prefect's engine handles the transition between synchronous and asynchronous contexts to prevent deadlocks and ensure that the event loop remains responsive for other asynchronous operations [1][3][2]. Specifically: - Synchronous tasks and flows are treated as "synchronous bridges" in the engine, meaning they run in a way that avoids introducing new event loops or threads unnecessarily [3]. - When calling a synchronous task or flow from an asynchronous one, Prefect does not simply block the event loop; it uses internal mechanisms to allow the synchronous code to run while maintaining the integrity of the existing event loop [1][2]. - In cases where synchronous tasks are submitted via submit() (e.g., using the default ThreadPoolTaskRunner), they are typically executed in separate worker threads, keeping them distinct from the main thread's event loop [4]. Therefore, while synchronous code in Prefect 3.6 interacts with the asynchronous engine, it does not "run" as a coroutine inside the event loop in the traditional sense; rather, it is handled by specialized bridges that manage the execution context to prevent performance degradation or deadlocks [1][3][2].
Citations:
- 1: Run tasks on the main thread PrefectHQ/prefect#11930
- 2: Run tasks on the main thread PrefectHQ/prefect#9855
- 3: New engine: add dedicated synchronous function handling PrefectHQ/prefect#12889
- 4: https://docs.prefect.io/v3/how-to-guides/workflows/write-and-run
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Files:"
git ls-files | rg '(^|/)prefect_flows_runner\.py$|($|/)prefect_flows\.py$|($|/)pyproject\.toml$|($|/)requirements.*\.txt$|setup.*\.py$|Pipfile|poetry.lock|uv.lock' || true
echo
echo "Relevant snippets:"
if [ -f proxy/prefect_flows_runner.py ]; then
nl -ba proxy/prefect_flows_runner.py | sed -n '520,615p'
fi
echo
echo "Search deployment_schedule_flow_v5:"
rg -n "def deployment_schedule_flow_v5|deployment_schedule_flow_v5|`@flow`" proxy -S || true
echo
echo "Task runner calls:"
rg -n "_run_task_runner|deployment_schedule_flow_v5|dbtcloudjob_v1|run_refresh_schema_flow|greenback|asyncio.run|asyncio\\.get_running_loop" proxy -S || true
echo
echo "Prefect dependency versions:"
for f in $(git ls-files | rg '(^|/)pyproject\.toml$|(^|/)requirements.*\.txt$|(^|/)Pipfile$|poetry.lock$|uv.lock$'); do
echo "--- $f"
sed -n '1,220p' "$f" | rg -n 'prefect|greenback|python' || true
doneRepository: DalgoT4D/prefect-proxy
Length of output: 289
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "Relevant snippets in proxy/prefect_flows_runner.py:"
wc -l proxy/prefect_flows_runner.py
python3 - <<'PY'
with open('proxy/prefect_flows_runner.py', encoding='utf-8') as f:
for i, line in enumerate(f, 1):
if 520 <= i <= 620:
print(f"{i:5d}: {line.rstrip()}")
PY
echo
echo "Search for deployment_schedule_flow_v5, task runner, asyncio.run, Prefect/greenback:"
rg -n "deployment_schedule_flow_v5|_run_task_runner|task_runner|dbtcloudjob_v1|run_refresh_schema_flow|run_airbyte_connection_flow_v1|asyncio\.run|asyncio\.get_running_loop|greenback|from prefect import .*Flow|`@flow`|prefect|greenback" -S . || true
echo
echo "Dependency files:"
for f in uv.lock pyproject.toml requirements*.txt Pipfile setup.py; do
[ -f "$f" ] && { echo "--- $f"; sed -n '1,240p' "$f" | rg -n 'prefect|greenback|python' || true; }
doneRepository: DalgoT4D/prefect-proxy
Length of output: 50378
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="$(mktemp -d)"
python3 - <<'PY' "$tmp"
import urllib.request, tarfile, shutil, sys
base, out = sys.argv[1], sys.argv[2]
url = "https://files.pythonhosted.org/packages/13/35/d61322b6b24d95346d6a3ad924403581ff6c4c0f22b8fa29287d59710d40/prefect-3.6.29-py3-none-any.whl"
print("download", url)
urllib.request.urlretrieve(url, f"{out}/prefect-3.6.29-py3-none-any.whl")
print("downloaded")
PY
cd "$tmp"
unzip -q prefect-3.6.29-py3-none-any.whl
echo "tracked sync-flow / execution symbols:"
rg -n "greenback|get_running_loop|current_context|run_coroutine_threadsafe|asyncio\.run|start_task|run_flow|synchronous|synchronicity|ThreadPool" prefect -S || true
echo
echo "candidate flow execution files:"
git ls-files -z prefect | tr '\0' '\n' | rg '/(flow|task|runner|worker|orch|state|context|engine|greenback|asyncio)\.py$' || trueRepository: DalgoT4D/prefect-proxy
Length of output: 269
🏁 Script executed:
#!/bin/bash
set -euo pipefail
tmp="$(mktemp -d)"
python3 - "$tmp" <<'PY'
import urllib.request, sys
import os
try:
from https_proxy import ProxyConfig
raise SystemExit
except Exception:
pass
out = sys.argv[1]
os.makedirs(out, exist_ok=True)
url = "https://files.pythonhosted.org/packages/90/11/df2cd0c702d4180b2c3d44b7bf3fdebfdc67f108926ba8f5f90c9a2ad88f/prefect-3.6.29.tar.gz"
tmp, _ = urllib.request.urlretrieve(url)
out_tar = os.path.join(out, "prefect-3.6.29.tar.gz")
with open(tmp, "rb") as src, open(out_tar, "wb") as dst:
dst.write(src.read())
print(out_tar)
PY
mkdir -p "$tmp/py"
tar -xzf "$tmp/prefect-3.6.29.tar.gz" -C "$tmp/py"
echo "tracked sync-flow / execution symbols in unpacked source:"
rg -n "greenback|get_running_loop|asyncio\.get_running_loop|run_coroutine_threadsafe|asyncio\.run|run_coroutine|current_context|current_context_contextvar|start_task|run_flow|TaskRunner|ThreadPool|GreenThread|synchronicity|synchronous|async_run|run_.*flow" "$tmp/py/prefect-3.6.29" -S || trueRepository: DalgoT4D/prefect-proxy
Length of output: 2576
🌐 Web query:
Prefect 3.6.29 greenback synchronize_flow async_run running event loop asyncio.run()
💡 Result:
In Prefect, greenback is a dependency used to resolve deadlocks and performance issues when calling synchronous tasks from asynchronous flows [1][2]. It allows Prefect to execute synchronous tasks without blocking the event loop [1][2]. When you run an asynchronous Prefect flow, the engine manages an event loop [3]. If you attempt to call asyncio.run() inside an existing asynchronous flow, you will typically encounter an error because asyncio.run() attempts to create a new event loop, but only one event loop can run in a thread at a time. If you need to execute asynchronous code from within a flow, you should: 1. Use await for any asynchronous operations instead of creating a new event loop with asyncio.run() [4][3]. 2. If you need to run synchronous code that performs I/O, rely on Prefect's handling of these tasks [1][2]. Prefect uses greenback to create a shim that permits synchronous tasks to execute without blocking the main event loop [1][2]. 3. For parallel execution within an async flow, utilize asyncio.gather() or AnyIO task groups, rather than launching new event loops [5]. If you are encountering issues with synchronize_flow or async_run in the context of event loop management, ensure you are not manually managing the loop inside your flow functions. Prefect is designed to handle the event loop lifecycle automatically for you [3]. If you are working with legacy synchronous code that must run as part of an async flow, Prefect's infrastructure generally handles this transition automatically via the greenback dependency [1][2].
Citations:
- 1: Run tasks on the main thread PrefectHQ/prefect#11930
- 2: Run tasks on the main thread PrefectHQ/prefect#9855
- 3: https://docs.prefect.io/v3/concepts/flows
- 4: https://docs.prefect.io/v3/get-started
- 5: https://linen.prefect.io/t/16065823/ulva73b9p-can-i-run-asyncio-run-async-function-in-a-prefect-
Avoid nested asyncio.run() calls in Prefect flow execution
_run_task_runner is called from deployment_schedule_flow_v5, and run_refresh_schema_flow calls it transitively, so the DBT Cloud branch can become deployment_schedule_flow_v5 → _run_task_runner(...) → asyncio.run(dbtcloudjob_v1(...)). In Prefect 3.6 flows that already run on a Prefect-managed event loop, this can raise RuntimeError: asyncio.run() cannot be called from a running event loop. Use a Prefect-bridged async executor or refactor the async task to be awaited within the flow instead of starting a nested event loop.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@proxy/prefect_flows_runner.py` around lines 568 - 605, Update the DBTCLOUD
branch of _run_task_runner to avoid calling asyncio.run for dbtcloudjob_v1 when
invoked from Prefect-managed flows. Use the existing Prefect-compatible async
execution mechanism, or refactor the call chain so the coroutine is awaited
within deployment_schedule_flow_v5 while preserving the current task dispatch
behavior.
| # TODO: revert branch back to "main" once feature/prefect-secret-blk merges. | ||
| # prefect_flows_runner.py + deployment_schedule_flow_v5 only exist on the | ||
| # feature branch until then. | ||
| source = GitRepository( | ||
| url="https://github.com/DalgoT4D/prefect-proxy.git", | ||
| branch="feature/prefect-secret-blk", | ||
| ) | ||
| deployment_id = flow.from_source( | ||
| source=source, entrypoint="proxy/prefect_flows.py:deployment_schedule_flow_v4" | ||
| source=source, | ||
| entrypoint="proxy/prefect_flows_runner.py:deployment_schedule_flow_v5", |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Do not create production deployments from the temporary feature branch.
After this PR merges, every newly created deployment will still pull feature/prefect-secret-blk, rather than the merged source. Deployments will miss subsequent fixes and fail if that branch is removed. Switch this to main (or an intentionally pinned release revision) before merging.
Proposed fix
- # TODO: revert branch back to "main" once feature/prefect-secret-blk merges.
- # prefect_flows_runner.py + deployment_schedule_flow_v5 only exist on the
- # feature branch until then.
source = GitRepository(
url="https://github.com/DalgoT4D/prefect-proxy.git",
- branch="feature/prefect-secret-blk",
+ branch="main",
)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| # TODO: revert branch back to "main" once feature/prefect-secret-blk merges. | |
| # prefect_flows_runner.py + deployment_schedule_flow_v5 only exist on the | |
| # feature branch until then. | |
| source = GitRepository( | |
| url="https://github.com/DalgoT4D/prefect-proxy.git", | |
| branch="feature/prefect-secret-blk", | |
| ) | |
| deployment_id = flow.from_source( | |
| source=source, entrypoint="proxy/prefect_flows.py:deployment_schedule_flow_v4" | |
| source=source, | |
| entrypoint="proxy/prefect_flows_runner.py:deployment_schedule_flow_v5", | |
| source = GitRepository( | |
| url="https://github.com/DalgoT4D/prefect-proxy.git", | |
| branch="main", | |
| ) | |
| deployment_id = flow.from_source( | |
| source=source, | |
| entrypoint="proxy/prefect_flows_runner.py:deployment_schedule_flow_v5", |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@proxy/service.py` around lines 678 - 687, Update the GitRepository
configuration in the deployment creation flow around flow.from_source to use the
production main branch instead of the temporary feature/prefect-secret-blk
branch. Ensure newly created deployments pull the merged source; use an
intentionally pinned release revision only if that is the project’s established
deployment convention.
2 participants
Summary by CodeRabbit
New Features
Bug Fixes
Tests