https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> Prefect 3.4.21 broke deployment param...
# ask-community
a
@Marvin Prefect 3.4.21 broke deployment parameters for me. I created an issue on Github as well, but posting it here because I suspect I might be doing something wrong as no one has pointed it out. etl.py
Copy code
from datetime import date

from workflows.flows.etl import extraction_and_transformation

if __name__ == "__main__":
    extraction_and_transformation.serve(
        name="Extraction Transformation Loading",
        parameters={
            "source_name": "ABC",
            "database_export_date": date(2025, 8, 27),
            "bucket_name": "data-migration",
        },
    )
workflows/flows/etl.py
Copy code
@flow(name="Data Extraction and Transformation")
def extraction_and_transformation(
    source_name: str, database_export_date: date, bucket_name: str
):
    from prefect.runtime import flow_run, deployment, task_run

    print("prefect version", prefect.__version__)
    print("Deployment Parameters", deployment.parameters)
    print("Flow Run Parameters", flow_run.parameters)
    print("Task Run Parameters", task_run.parameters)
    print("Flow Run Parent Deployment ID", flow_run.parent_deployment_id)
Copy code
prefect version 3.4.20
Deployment Parameters {'bucket_name': 'data-migration', 'source_name': 'ABC', 'database_export_date': '2025-08-27'}
Flow Run Parameters {'source_name': 'ABC', 'database_export_date': datetime.date(2025, 8, 27), 'bucket_name': 'data-migration'}
Task Run Parameters {}
Copy code
prefect version 3.4.21
Deployment Parameters {}
Flow Run Parameters {'source_name': 'ABC', 'database_export_date': datetime.date(2025, 8, 27), 'bucket_name': 'data-migration'}
Task Run Parameters {}
m
✅ thought for 284.0 seconds
Thanks for the clear repro — that’s really helpful. Short answer - You’re not doing anything wrong. In 3.4.21, 
prefect.runtime.deployment.parameters
being 
{}
in this scenario looks like a behavioral change/regression. - The parameters your flow actually runs with are correctly available at 
prefect.runtime.flow_run.parameters
— that’s the reliable place to read them at runtime. - In 3.4.20, 
deployment.parameters
happened to include the values you passed to `.serve(parameters=...)`; in 3.4.21 it’s empty. I can’t find any release note or discussion that documents this change, so I suspect a regression in how the deployment runtime context is populated for 
.serve()
. What’s the difference? - 
prefect.runtime.flow_run.parameters
= the effective parameters for the current flow run (what your code should use at runtime). - 
prefect.runtime.deployment.parameters
= parameters configured on the deployment object itself (deployment defaults). These are JSON-serialized; that’s why your date showed up as a string in 3.4.20. Workarounds - Prefer this inside your flow when you need the parameters your run is using:
Copy code
from prefect.runtime import flow_run
  params = flow_run.parameters
- If you truly need the deployment’s default parameters (as stored on the deployment object), fetch them via the client using the deployment id:
Copy code
import anyio
  from prefect.client import get_client
  from prefect.runtime import deployment

  async def get_deployment_params():
      async with get_client() as client:
          dep = await client.read_deployment(deployment.id)
          return dep.parameters

  params = anyio.run(get_deployment_params)
  print(params)
- If your logic relied on 
deployment.parameters
being populated by 
.serve(parameters=...)
, consider pinning Prefect to 3.4.20 until this is clarified/fixed:
Copy code
pip install "prefect==3.4.20"
Questions to help us confirm scope - In 3.4.21, is 
prefect.runtime.deployment.id
populated for the same run? - Are you on Prefect Cloud or OSS? - Could you share the GitHub issue link here so we can track it and reference it for others? Related docs - Runtime context: Runtime context - `prefect.runtime.deployment`: prefect.runtime.deployment - `prefect.runtime.flow_run`: prefect.runtime.flow_run If you can share your issue link, I’ll keep an eye on it and update you here with any fixes or workarounds.
a
n
hi @Ahmad Bilal Khalid - thanks for reporting! taking a look at this
👍 1
a
Thanks @Nate, I will test in my morning and report back.
When will the new release be available?
n
we usually release on thursday, but you can install from 
main
at any time if you need/can
👍 1
a
@Nate This fix partially solves it, the parameters are now available in the flow but not in the tasks, previously they were available in the tasks as well. I tested it by installing 
prefect[ray,redis,shell]==3.5.1.dev1
n
hm - will take a look, feel free to share a reproduction if you have it though
👍 1
a
Copy code
Flow: prefect version 3.4.20
Flow: Deployment Parameters {'bucket_name': 'data-migration', 'export_date': '2025-08-27', 'source_name': 'ABC'}
Flow: Flow Run Parameters {'source_name': 'ABC', 'export_date': datetime.date(2025, 8, 27), 'bucket_name': 'data-migration'}
Flow: Task Run Parameters {}
Flow: Flow Run Parent Deployment ID None

Nested Flow: prefect version 3.4.20
Nested Flow: Deployment Parameters {'bucket_name': 'data-migration', 'export_date': '2025-08-27', 'source_name': 'ABC'}
Nested Flow: Flow Run Parameters {}
Nested Flow: Task Run Parameters {}
Nested Flow: Flow Run Parent Deployment ID b037183c-e175-4cdb-9bef-d3f4817c2a6c

(a_task pid=1077) Task: prefect version 3.4.20
(a_task pid=1077) Task: Deployment Parameters {'bucket_name': 'data-migration', 'export_date': '2025-08-27', 'source_name': 'ABC'}
(a_task pid=1077) Task: Flow Run Parameters {}
(a_task pid=1077) Task: Task Run Parameters {}
(a_task pid=1077) Task: Flow Run Parent Deployment ID b037183c-e175-4cdb-9bef-d3f4817c2a6c
Copy code
Flow: prefect version 3.5.1.dev1
Flow: Deployment Parameters {'bucket_name': 'data-migration', 'export_date': '2025-08-27', 'source_name': 'ABC'}
Flow: Flow Run Parameters {'source_name': 'ABC', 'export_date': datetime.date(2025, 8, 27), 'bucket_name': 'data-migration'}
Flow: Task Run Parameters {}
Flow: Flow Run Parent Deployment ID None

Nested Flow: prefect version 3.5.1.dev1
Nested Flow: Deployment Parameters {}
Nested Flow: Flow Run Parameters {}
Nested Flow: Task Run Parameters {}
Nested Flow: Flow Run Parent Deployment ID b037183c-e175-4cdb-9bef-d3f4817c2a6c

(a_task pid=805) Task: prefect version 3.5.1.dev1
(a_task pid=805) Task: Deployment Parameters {}
(a_task pid=805) Task: Flow Run Parameters {}
(a_task pid=805) Task: Task Run Parameters {}
Copy code
from prefect import flow, task
import prefect
from datetime import date

from prefect_ray import RayTaskRunner


def get_task_runner():
    RAY_INIT_KWARGS = {
        "num_cpus": 5,
        "runtime_env": {
            "excludes": [
                "*.pyc",
                "*.pyo",
                ".ruff_cache",
                ".venv",
                "uv.lock",
                ".git",
                "__pycache__",
            ]
        },
        # 75 MB of object store memory, we don't use it
        "object_store_memory": 75 * 1024 * 1024,
    }
    return RayTaskRunner(init_kwargs=RAY_INIT_KWARGS)


@task()
def a_task():
    from prefect.runtime import flow_run, deployment, task_run

    print("Task: prefect version", prefect.__version__)
    print("Task: Deployment Parameters", deployment.parameters)
    print("Task: Flow Run Parameters", flow_run.parameters)
    print("Task: Task Run Parameters", task_run.parameters)
    print("Task: Flow Run Parent Deployment ID", flow_run.parent_deployment_id)


@flow(name="Nested Flow", task_runner=get_task_runner())
def nested_flow():
    from prefect.runtime import flow_run, deployment, task_run

    print("Nested Flow: prefect version", prefect.__version__)
    print("Nested Flow: Deployment Parameters", deployment.parameters)
    print("Nested Flow: Flow Run Parameters", flow_run.parameters)
    print("Nested Flow: Task Run Parameters", task_run.parameters)
    print("Nested Flow: Flow Run Parent Deployment ID", flow_run.parent_deployment_id)

    a_task.submit().result()


@flow(name="Data Extraction and Transformation")
def extraction_and_transformation(
    source_name: str, export_date: date, bucket_name: str
):
    from prefect.runtime import flow_run, deployment, task_run

    print("Flow: prefect version", prefect.__version__)
    print("Flow: Deployment Parameters", deployment.parameters)
    print("Flow: Flow Run Parameters", flow_run.parameters)
    print("Flow: Task Run Parameters", task_run.parameters)
    print("Flow: Flow Run Parent Deployment ID", flow_run.parent_deployment_id)

    return nested_flow()


if __name__ == "__main__":
    extraction_and_transformation.serve(
        name="Extraction Transformation Loading",
        parameters={
            "source_name": "ABC",
            "export_date": "2025-08-27",
            "bucket_name": "data-migration",
        },
    )
@Nate I have added the example
@Nate Thanks nate for quick fix, although parameters are still missing in tasks, I added details on Github PR.
n
hm! thanks for taking a look! will update there
👍 1
ah i see, it works for normal tasks but the serialization logic for multi-process task runners was not updated accordingly
ill update the PR, thanks for catching this!
a
@Nate I tested it and it worked perfectly. My workflow is green again, thanks to you. 👍🏼
n
catjam
thanks for testing!
👍 1
Open in Slack
PreviousNext
Powered by