Hey folks a question about prefect 2 0 and pydantic I certai

Question

Hey folks a question about prefect 2 0 and pydantic I certainly appreciate using pydantic at the flow level to allow for complex parameter types and to enable type validation conversion out of the box However it seems that pydantic supports starts to fail at the task level More specifically it seems pickling pydantic models using cloudpickle is troublesome See the thread for a small example I am curious if this will remain a known issue with certain storage types going forward

Marwan Sarieddine · Answer

Here is the flow code I took the code and added one additional task ```from prefect import flow task get run logger from pydantic import BaseModel class Model BaseModel a int b float c str task def parser obj Model gt Model logger = get run logger f obj a= obj b= obj c= return obj task def printer obj Model gt None logger = get run logger f Received a type obj with value obj flow def model validator model Model gt None obj = parser model printer obj=obj model validator a 42 b 0 c 55 ```

Marwan Sarieddine · Answer

if I run this I get this error ```16 32 15 606 | INFO | prefect engine Created flow run sensible nuthatch for flow model validator 16 32 15 789 | INFO | Flow run sensible nuthatch Created task run parser dda0fc11 0 for task parser 16 32 15 790 | INFO | Flow run sensible nuthatch Executing parser dda0fc11 0 immediately 16 32 15 819 | INFO | Task run parser dda0fc11 0 obj a=42 obj b=0 0 obj c= 55 16 32 15 820 | ERROR | Flow run sensible nuthatch Encountered exception during execution Traceback most recent call last File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 557 in orchestrate flow run result = await run sync flow call File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect utilities asyncutils py line 56 in run sync in worker thread return await anyio to thread run sync call cancellable=True File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 867 in run result = context run func args File Users marwansarieddine demo interests orion orion flow 2 py line 23 in model validator obj = parser model File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect tasks py line 294 in call return enter task run engine File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 687 in enter task run engine return run async from worker thread begin run File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect utilities asyncutils py line 136 in run async from worker thread return anyio from thread run call File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio from thread py line 49 in run return asynclib run async from thread func args File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 970 in run async from thread return f result File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 concurrent futures base py line 446 in result return self get result File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 concurrent futures base py line 391 in get result raise self exception File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 806 in create task run then submit return await future result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect futures py line 220 in result return final state result raise on failure=raise on failure File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect orion schemas states py line 145 in result raise data File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect task runners py line 214 in submit result = await run fn run kwargs File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 947 in begin task run return await orchestrate task run File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 1065 in orchestrate task run terminal state = await return value to state File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect states py line 129 in return value to state return Completed data=DataDocument encode serializer result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect orion schemas data py line 42 in encode blob = lookup serializer encoding dumps data kwargs File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect serializers py line 59 in dumps data bytes = cloudpickle dumps data File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages cloudpickle cloudpickle fast py line 73 in dumps cp dump obj File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages cloudpickle cloudpickle fast py line 633 in dump return Pickler dump self obj pickle PicklingError Can t pickle lt cyfunction int validator at 0x1079a91e0 gt it s not the same object as pydantic validators int validator 16 32 15 828 | INFO | Task run parser dda0fc11 0 Crash detected Execution was interrupted by an unexpected exception 16 32 15 883 | ERROR | Flow run sensible nuthatch Finished in state Failed Flow run encountered an exception Traceback most recent call last File Users marwansarieddine demo interests orion orion flow 2 py line 26 in lt module gt model validator a 42 b 0 c 55 File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect flows py line 390 in call return enter flow run engine from flow call File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 152 in enter flow run engine from flow call return anyio run begin run File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio core eventloop py line 70 in run return asynclib run func args backend options File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 292 in run return native run wrapper debug=debug File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 asyncio runners py line 44 in run return loop run until complete main File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 asyncio base events py line 647 in run until complete return future result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 287 in wrapper return await func args File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect client py line 104 in with injected client return await fn args kwargs File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 228 in create then begin flow run return state result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect orion schemas states py line 145 in result raise data File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 557 in orchestrate flow run result = await run sync flow call File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect utilities asyncutils py line 56 in run sync in worker thread return await anyio to thread run sync call cancellable=True File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio to thread py line 31 in run sync return await get asynclib run sync in worker thread File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 937 in run sync in worker thread return await future File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 867 in run result = context run func args File Users marwansarieddine demo interests orion orion flow 2 py line 23 in model validator obj = parser model File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect tasks py line 294 in call return enter task run engine File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 687 in enter task run engine return run async from worker thread begin run File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect utilities asyncutils py line 136 in run async from worker thread return anyio from thread run call File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio from thread py line 49 in run return asynclib run async from thread func args File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages anyio backends asyncio py line 970 in run async from thread return f result File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 concurrent futures base py line 446 in result return self get result File Users marwansarieddine pyenv versions 3 9 12 lib python3 9 concurrent futures base py line 391 in get result raise self exception File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 806 in create task run then submit return await future result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect futures py line 220 in result return final state result raise on failure=raise on failure File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect orion schemas states py line 145 in result raise data File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect task runners py line 214 in submit result = await run fn run kwargs File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 947 in begin task run return await orchestrate task run File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect engine py line 1065 in orchestrate task run terminal state = await return value to state File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect states py line 129 in return value to state return Completed data=DataDocument encode serializer result File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect orion schemas data py line 42 in encode blob = lookup serializer encoding dumps data kwargs File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages prefect serializers py line 59 in dumps data bytes = cloudpickle dumps data File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages cloudpickle cloudpickle fast py line 73 in dumps cp dump obj File Users marwansarieddine pyenv versions prefect2 lib python3 9 site packages cloudpickle cloudpickle fast py line 633 in dump return Pickler dump self obj pickle PicklingError Can t pickle lt cyfunction int validator at 0x1079a91e0 gt it s not the same object as pydantic validators int validator ```

Anna Geller · Answer

you can pass validate parameters=False to disable validation

Marwan Sarieddine · Answer

So I saw this but it is problematic for more than one reason 1 It does not resolve the error given the input now remains a dictionary and doesn t get converted into a pydantic Model 2 We lose the parameter validation at the flow level

Marwan Sarieddine · Answer

Is the suggestion that if one wants use tasks and a given storage type that relies on cloudpickle that for now pydantic validation is not supported

Anna Geller · Answer

we don t have pydantic validation on a task level

Anna Geller · Answer

that must be major source of confusion for you

Anna Geller · Answer

we do it only for flows to validate input parameters to the flow

Anna Geller · Answer

feel free to open a GitHub issue with a feature request to add task level validation and we could discuss in a team but this won t work atm

Marwan Sarieddine · Answer

no I understand there is no pydantic validation on the task level that is not what I am requesting I would like to have flow level validation with pydantic and to also use tasks in my flow

Anna Geller · Answer

gt a given storage type that relies on cloudpickle we deprecated cloudpickle for flow storage too

Marwan Sarieddine · Answer

gt I would like to have flow level validation with pydantic and to also use tasks in my flow As you see from the example I shared this raises an error I am on the latest prefect version

Anna Geller · Answer

ok sorry let me investigate a bit more before telling you sth wrong

Anna Geller · Answer

ok the thing is you can totally use pydantic models in your tasks and call such tasks in your flow the problem is with returning a pydantic model e g changing your task this way will work ``` task def parser obj Model gt Model logger = get run logger <http logger info|logger info> f obj a= obj b= obj c= return obj```

Anna Geller · Answer

do you have to return it in your use case if not this should fix your issue

Marwan Sarieddine · Answer

yea so I can t pass pydantic Model instances between tasks this is unfortunate

Marwan Sarieddine · Answer

thanks for confirming this is the case thought I was missing something

Anna Geller · Answer

do you need this functionality

Marwan Sarieddine · Answer

potentially yes still in the early phases of exploring migrating to prefect 2

Anna Geller · Answer

this is more sth related to pydantic than to Prefect <https github com cloudpipe cloudpickle issues 408>

Anna Geller · Answer

I believe you shouldn t worry too much about this since Michael is currently working on making the task run persistence more configurable so I believe soon you should be able to avoid checkpointing pickling such pydantic objects and thus you should be able to pass it between tasks

Anna Geller · Answer

making task run results persistent more configurable is a high priority roadmap item atm so I would expect to see this being tackled within the next 2 weeks

Marwan Sarieddine · Answer

ok will keep an eye out for this update thanks for providing the support and update much appreciated

Felix Sonntag · Answer

Also +1 skin tone 2 for that currently facing issues with serializing Keras models slightly smiling face

Anna Geller · Answer

Thanks to you both for proving the urgency of this I ll ping the team to highlight the importance of this issue to prioritize

Oliver Mannion · Answer

Thank you for the link I ran into this and resolved it by moving my model into a separate file from my flow as per <https github com cloudpipe cloudpickle issues 408 issuecomment 933760919|this comment>