https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i have one workflow using python 3.12...
# ask-marvin
o
@Marvin i have one workflow using python 3.12 and one using python 3.11 also i have one workflow triggereing both. how do i do that?
m
✅ thought for 246.7 seconds
Great question — you can do this cleanly in Prefect 3.x by giving each workflow its own deployment bound to a work pool whose workers run the required Python version. Then have a third “orchestrator” flow trigger those deployments. High level - Create two deployments, one per workflow: - Workflow A → Python 3.11 - Workflow B → Python 3.12 - Point each deployment at a different work pool with workers that run the correct Python. - In your orchestrator flow, call 
run_deployment("FlowName/DeploymentName", ...)
for each. Two common ways to ensure the right Python per workflow Option A — Docker work pools (recommended for strong isolation) 1) Create two Docker work pools:
Copy code
prefect work-pool create py311-pool --type docker
prefect work-pool create py312-pool --type docker
Note: install the extra if needed: 
pip install 'prefect[docker]'
. 2) Build/push two images that contain your code/deps for each Python version (e.g., 
<http://ghcr.io/acme/app:py311|ghcr.io/acme/app:py311>
, `ghcr.io/acme/app:py312`). 3) Deploy each flow to the matching pool and image. With the Python API:
Copy code
from prefect import flow

@flow
def flow_py311(...):
    ...

@flow
def flow_py312(...):
    ...

# Deploy the 3.11 flow
flow_py311.from_source(
    "<https://github.com/your-org/your-repo>",
    entrypoint="flows/py311_flow.py:flow_py311",
).deploy(
    name="prod",
    work_pool_name="py311-pool",
    job_variables={"image": "<http://ghcr.io/acme/app:py311|ghcr.io/acme/app:py311>"},
)

# Deploy the 3.12 flow
flow_py312.from_source(
    "<https://github.com/your-org/your-repo>",
    entrypoint="flows/py312_flow.py:flow_py312",
).deploy(
    name="prod",
    work_pool_name="py312-pool",
    job_variables={"image": "<http://ghcr.io/acme/app:py312|ghcr.io/acme/app:py312>"},
)
You can also use the CLI from each project root:
Copy code
prefect deploy ./flows/py311_flow.py:flow_py311 --name "flow_py311/prod" --pool py311-pool --job-variable image=<http://ghcr.io/acme/app:py311|ghcr.io/acme/app:py311>
prefect deploy ./flows/py312_flow.py:flow_py312 --name "flow_py312/prod" --pool py312-pool --job-variable image=<http://ghcr.io/acme/app:py312|ghcr.io/acme/app:py312>
4) Start one worker per pool (e.g., on machines with Docker):
Copy code
prefect worker start -p py311-pool
prefect worker start -p py312-pool
Option B — Process workers + venvs (simple local isolation) - Create two virtual environments: - venv311 with Python 3.11 + deps - venv312 with Python 3.12 + deps - Create two process work pools:
Copy code
prefect work-pool create py311-pool --type process
prefect work-pool create py312-pool --type process
- From each venv, start a worker for its pool: - In venv311: 
prefect worker start -p py311-pool
- In venv312: 
prefect worker start -p py312-pool
- Deploy each flow to its matching pool (similar to the Docker example, just omit 
job_variables.image
). Orchestrator flow that triggers both Use 
run_deployment
to kick off the two deployments. You can wait for them to finish or fire-and-forget. Wait for both to finish (concurrently):
Copy code
from prefect import flow
from prefect.deployments import run_deployment
import asyncio

@flow
async def orchestrator():
    r1, r2 = await asyncio.gather(
        run_deployment("flow_py311/prod", parameters={"x": 1}),
        run_deployment("flow_py312/prod", parameters={"y": 2}),
    )
    return {"py311_state": r1.state.name, "py312_state": r2.state.name}
Fire-and-forget (return immediately after submitting the runs):
Copy code
@flow
async def orchestrator_fire_and_forget():
    r1 = await run_deployment("flow_py311/prod", parameters={"x": 1}, timeout=0)
    r2 = await run_deployment("flow_py312/prod", parameters={"y": 2}, timeout=0)
    return {"py311_id": r1.id, "py312_id": r2.id}
Notes and tips - 
run_deployment
name is 
"flow_name/deployment_name"
. It returns a 
FlowRun
. - By default, 
as_subflow=True
, so the triggered runs appear as subflows under your orchestrator in the UI (recommended). - Waiting behavior: - 
timeout=None
waits indefinitely for completion. - 
timeout=0
returns immediately (fire-and-forget).
- 
poll_interval
controls how often completion is checked while waiting (default 5s). Helpful docs - Run deployments programmatically: Run deployments - API reference for `run_deployment`: prefect.deployments.flow_runs.run_deployment - Deployments and work pools overview: Deployments concept Questions to tailor this for you - Which infrastructure do you want to run on (local process, Docker, Kubernetes, something else)? - Do you want the orchestrator to wait for both workflows to finish, or just trigger them and move on? - Are your flows in one repo or separate repos? And do you already have Docker images or prefer venvs? If you share those details, I can provide a concrete, cut-and-paste config for your setup.
o
@Marvin what about this error? could it be that i cant input the image? must i serialize it? Encountered exception during execution: PydanticSerializationError(Unable to serialize unknown type: <class 'numpy.ndarray'>) Traceback (most recent call last): File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 1357, in run_context yield self File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 1419, in run_flow_async await engine.call_flow_fn() File "/usr/local/lib/python3.12/site-packages/prefect/flow_engine.py", line 1371, in call_flow_fn result = await call_with_parameters(self.flow.fn, self.parameters) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/app/deployments_and_subdeployment/concurrent_execution.py", line 19, in a_flow_calls_two_subflows_already_deployed return1, return2 = await asyncio.gather( ^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/utilities/asyncutils.py", line 341, in ctx_call result = await async_fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/client/utilities.py", line 99, in with_injected_client return await fn(*args, **kwargs) ^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/deployments/flow_runs.py", line 183, in run_deployment flow_run = await client.create_flow_run_from_deployment( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/prefect/client/orchestration/_deployments/client.py", line 1360, in create_flow_run_from_deployment json=flow_run_create.model_dump(mode="json", exclude_unset=True), ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ File "/usr/local/lib/python3.12/site-packages/pydantic/main.py", line 463, in model_dump return self.__pydantic_serializer__.to_python( ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ pydantic_core._pydantic_core.PydanticSerializationError: Unable to serialize unknown type: <class 'numpy.ndarray'> 032847 PM prefect.flow_runs Finished in state Failed("Flow run encountered an exception: PydanticSerializationError: Unable to serialize unknown type: <class 'numpy.ndarray'>") 032847 PM prefect.flow_runs Process for flow run 'crystal-bug' exited with status code: 1
m
✅ thought for 115.1 seconds
You’re hitting that error because a numpy ndarray is being sent as a flow parameter. Prefect 3.x serializes parameters to JSON when creating the flow run; numpy arrays aren’t JSON-serializable, so Pydantic fails. Important distinction: - Docker image for the worker: pass this as 
job_variables={"image": "repo:tag"}
— that’s a plain string and totally fine. - Image data (e.g., a numpy array): don’t pass this directly in `parameters`; it must be serialized or referenced. Ways to fix it 1) Prefer passing a reference (best for large data) - Upload the image to storage (S3/GCS/Azure/local path) and pass a URI/path/id.
Copy code
await run_deployment(
  "flow_py311/prod",
  parameters={"image_uri": "<s3://my-bucket/path/to/image.png>"},
  job_variables={"image": "<http://ghcr.io/acme/app:py311|ghcr.io/acme/app:py311>"}
)
Then the child flow loads the image from the URI. 2) If the array is small, convert to JSON-friendly types - Convert to lists (can be large/slow for big arrays):
Copy code
parameters={"image_array": image_np.tolist()}
3) Serialize to a string (only if you must embed the data; watch size limits) - Use pickle + base64. Note: only unpickle data you trust.
Copy code
import base64, pickle

def encode_ndarray(arr):
    return base64.b64encode(pickle.dumps(arr)).decode("utf-8")

def decode_ndarray(s):
    return pickle.loads(base64.b64decode(s.encode("utf-8")))

# Orchestrator side
encoded = encode_ndarray(image_np)
await run_deployment(
  "flow_py312/prod",
  parameters={"image_b64_pickle": encoded},
  timeout=0,  # optional: fire-and-forget
)

# Child flow side
arr = decode_ndarray(image_b64_pickle)
Also, to set the Docker image for the worker use 
job_variables
, not a parameter:
Copy code
await run_deployment(
  "flow_py311/prod",
  parameters={"x": 1},
  job_variables={"image": "<http://ghcr.io/acme/app:py311|ghcr.io/acme/app:py311>"}
)
Docs for reference: - Run deployments: Run deployments - Serializers API (if you want to roll your own): prefect.serializers If you share the exact 
run_deployment(...)
call you’re using (especially what’s inside 
parameters
), I can suggest the smallest change to make it work.
6 Views
Open in Slack
PreviousNext
Powered by