Skip to content

Prefect client creates incorrect flow runs parameters due to unexpected serialization behavior #17887

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
levzem opened this issue Apr 22, 2025 · 2 comments · Fixed by #17938
Closed
Labels
bug Something isn't working

Comments

@levzem
Copy link
Contributor

levzem commented Apr 22, 2025

Bug summary

https://github.com/PrefectHQ/prefect/blob/main/src/prefect/client/orchestration/_deployments/client.py#L614

The Prefect client uses the exclude_unset setting for model_dump(), which has the unfortunate side effect of dropping default fields that were mutated afterwards.

A simple example is

class MyFlowParam(BaseModel):

    dropped: list[str] = Field(default_factory=list)

param = MyFlowParam()
param.dropped.append("mutated")

my_flow(param)

this will persist the flow run parameters as {"param": {}} instead of {"param": {"dropped": ["mutated"]}}

I would recommend considering setting exclude_default instead of exclude_unset as this is a really nasty foot gun and took a lot of time for me to get to the bottom of.

See for more explanations below:
pydantic/pydantic#5749
pydantic/pydantic#9866

Version info

prefect: 3.2.6

Additional context

No response

@levzem levzem added the bug Something isn't working label Apr 22, 2025
@zzstoatzz
Copy link
Collaborator

zzstoatzz commented Apr 29, 2025

@levzem thanks for the issue and the linked context

your example doesn't quite reproduce, but this does

from threading import Thread

from pydantic import BaseModel, Field

from prefect import flow
from prefect.deployments import run_deployment


class MyFlowParam(BaseModel):
    dropped: list[str] = Field(default_factory=list)


@flow
def my_flow(param: MyFlowParam):
    print(param)


if __name__ == "__main__":
    param = MyFlowParam()
    param.dropped.append("foo")

    Thread(target=my_flow.serve, daemon=True).start()

    run_deployment(
        "my-flow/my-flow",
        parameters={"param": param}, # works fine if you replace `param` with `param.model_dump()`
    )

looking more into this now! please feel free to add more detail on how you're actually hitting this in your usage of prefect

@levzem
Copy link
Contributor Author

levzem commented May 13, 2025

thanks for fixing this so promptly @zzstoatzz !

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
bug Something isn't working
Projects
None yet
Development

Successfully merging a pull request may close this issue.

2 participants