Hi everyone, could anyone help me setting a new custom serializer as the default? I have an extended version of the JSONSerializer, and I'd like to be automatically used by all of my tasks
Paul Gierz
02/05/2025, 3:46 PM
Specifically, I have something like this:
Copy code
class ProvenanceSerializer(JSONSerializer):
"""Serializes Provenance objects into JSON"""
type: str = Field(default="provenance_json", frozen=True)
def dumps(self, obj) -> bytes:
breakpoint()
if str(obj.__class__).endswith("WithProvenance"):
obj = obj._to_prefect_json()
return super().dumps(obj)
def loads(self, blob):
breakpoint()
data = super().loads(blob)
if "provenance" in data:
value = data["value"]
provenance = data["provenane"]
return wrapper_with_provenance_factory(value, provenance)
else:
return data
Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.