Add arun_deployment and replace @sync_compatible with @async_dispatch#173
Conversation
This change follows the intent of issue PrefectHQ#15008 to replace implicit sync/async conversion with explicit, type-safe alternatives. Changes: - Add `arun_deployment` as an explicit async function for running deployments - Replace `@sync_compatible` with `@async_dispatch` on `run_deployment` - `run_deployment` now dispatches to `arun_deployment` in async context - Sync context uses `SyncPrefectClient` directly (no event loop magic) - Export `arun_deployment` from `prefect.deployments` - Add comprehensive tests for both sync and async behavior The `run_deployment.aio` attribute is preserved for backward compatibility. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Review Summary by QodoReplace @sync_compatible with @async_dispatch for explicit async deployment handling
WalkthroughsDescription• Replace @sync_compatible with explicit @async_dispatch decorator on run_deployment • Add new arun_deployment async function for explicit async deployment execution • Implement proper client initialization using get_or_create_client instead of inject_client • Add comprehensive test suite for async, sync, and dispatch behavior • Create backward compatibility alias and .aio attribute for existing code Diagramflowchart LR
A["run_deployment<br/>@async_dispatch"] -->|"async context"| B["arun_deployment<br/>explicit async"]
A -->|"sync context"| C["SyncPrefectClient<br/>direct execution"]
D["Backward<br/>Compatibility"] -->|".aio attribute"| B
E["Client<br/>Initialization"] -->|"get_or_create_client"| F["PrefectClient<br/>instance"]
File Changes1. src/prefect/deployments/__init__.py
|
Code Review by Qodo
1. flow_runs.py missing future annotations
|
![qodo-code-review[bot]](https://avatars.githubusercontent.com/in/484649?s=60&v=4){kind=small}
Comment on lines
+11
to
+18
| from prefect.client.utilities import get_or_create_client | ||
| from prefect.context import FlowRunContext, TaskRunContext | ||
| from prefect.logging import get_logger | ||
| from prefect.states import Pending, Scheduled | ||
| from prefect.tasks import Task | ||
| from prefect.telemetry.run_telemetry import LABELS_TRACEPARENT_KEY, RunTelemetry | ||
| from prefect.types._datetime import now | ||
| from prefect.utilities.asyncutils import sync_compatible | ||
| from prefect.utilities._engine import dynamic_key_for_task_run |
There was a problem hiding this comment.
1. flow_runs.py missing future annotations 📘 Rule violation ✓ Correctness
src/prefect/deployments/flow_runs.py contains type annotations but does not include `from __future__ import annotations` as the first import. This can break forward references and violates the required typing import standard.
Agent Prompt
## Issue description
`src/prefect/deployments/flow_runs.py` uses type annotations but does not start with `from __future__ import annotations`, which is required for consistent forward-reference behavior.
## Issue Context
This file contains many annotations (e.g., `Optional["PrefectClient"]`, return type `"FlowRun"`). The compliance rule requires the future import as the first import in annotated `src/` files.
## Fix Focus Areas
- src/prefect/deployments/flow_runs.py[1-3]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+240
to
+241
| # Alias for backwards compatibility | ||
| run_deployment = arun_deployment |
There was a problem hiding this comment.
2. run_deployment lacks sync wrapper 📘 Rule violation ⛯ Reliability
run_deployment is now just an alias to the async arun_deployment, so there is no true sync-compatible wrapper. This violates the requirement that public async user-facing APIs provide sync compatibility.
Agent Prompt
## Issue description
`run_deployment` is currently just an alias to `arun_deployment` (`run_deployment = arun_deployment`), so it does not provide sync compatibility for user-facing callers.
## Issue Context
The compliance requirement expects public async APIs to offer sync wrappers (either via `@sync_compatible` or an explicit sync function). The codebase already uses `@async_dispatch(async_impl)` for this purpose.
## Fix Focus Areas
- src/prefect/deployments/flow_runs.py[48-61]
- src/prefect/deployments/flow_runs.py[240-241]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
| from prefect.client.schemas.responses import DeploymentResponse | ||
| from prefect.context import FlowRunContext | ||
| from prefect.deployments import run_deployment | ||
| from prefect.deployments import arun_deployment, run_deployment |
There was a problem hiding this comment.
3. Tests not in deployments/ 📘 Rule violation ⛯ Reliability
The tests for src/prefect/deployments/flow_runs.py are located under tests/deployment/ instead of mirroring the deployments/ module path. This violates the required test directory mirroring convention.
Agent Prompt
## Issue description
Tests for the `deployments/flow_runs.py` source module live under `tests/deployment/` instead of `tests/deployments/`, violating the required mirrored structure.
## Issue Context
The compliance rule expects a mirrored path for discoverability and consistency.
## Fix Focus Areas
- tests/deployment/test_flow_runs.py[1-40]
- src/prefect/deployments/flow_runs.py[1-5]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
211
to
213
| flow_run = await client.create_flow_run_from_deployment( | ||
| deployment.id, | ||
| deployment_id, | ||
| parameters=parameters, |
There was a problem hiding this comment.
4. None deployment_id passed 🐞 Bug ✓ Correctness
arun_deployment calls client.create_flow_run_from_deployment(deployment_id, ...) even when name is a non-UUID deployment name, leaving deployment_id=None and causing an invalid required UUID argument at runtime.
Agent Prompt
### Issue description
`arun_deployment` passes `deployment_id` into `create_flow_run_from_deployment`, but `deployment_id` remains `None` when `name` is a deployment name like `"flow/deployment"`. This breaks the primary documented use-case.
### Issue Context
You already load `deployment` via `read_deployment_by_name(name)` when `deployment_id` is not provided; the `deployment.id` should be used for creating the flow run.
### Fix Focus Areas
- src/prefect/deployments/flow_runs.py[123-139]
- src/prefect/deployments/flow_runs.py[211-222]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
146
to
150
|
|
||
| # This was called from a flow. Link the flow run as a subflow. | ||
| task_inputs = { | ||
| k: await collect_task_run_inputs(v) for k, v in parameters.items() | ||
| k: collect_task_run_inputs(v) for k, v in parameters.items() | ||
| } |
There was a problem hiding this comment.
5. Missing await task inputs 🐞 Bug ✓ Correctness
When linking a subflow, task_inputs uses collect_task_run_inputs(v) without await even though collect_task_run_inputs is async, resulting in coroutine objects being passed as task inputs.
Agent Prompt
### Issue description
The subflow-linking path populates `task_inputs` with coroutine objects because `collect_task_run_inputs` is async but is called without `await`.
### Issue Context
This code runs inside `arun_deployment` (async), so `await` is the correct way to obtain the expected `set[...]` values.
### Fix Focus Areas
- src/prefect/deployments/flow_runs.py[143-196]
- src/prefect/utilities/engine.py[66-115]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
Comment on lines
+133
to
+134
| client, _ = get_or_create_client(client) | ||
|
|
There was a problem hiding this comment.
6. Unclosed created client 🐞 Bug ⛯ Reliability
arun_deployment uses get_or_create_client but ignores the returned "inferred" flag and never context-manages/close the client when it had to create a new one.
Agent Prompt
### Issue description
When `client` is not provided and no context client exists, `get_or_create_client` creates a new client that should be closed. `arun_deployment` currently discards the inferred/created flag and never closes the created client.
### Issue Context
The previous `@inject_client` decorator handled this via `async with` when `inferred` is False.
### Fix Focus Areas
- src/prefect/deployments/flow_runs.py[48-139]
- src/prefect/client/utilities.py[31-61]
- src/prefect/client/utilities.py[74-100]
ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Benchmark PR from agentic-review-benchmarks#6