import prefect
from prefect import flow, task
from dataclasses import dataclass

@dataclass
class testDC:
    frodo: int
    baggings: int = None

    def __post_init__(self):
        print("post_init triggered")
        if self.baggings is None:
            print("self.baggings is None triggered")
            self.baggings = self.frodo * 2
        else:
            print("self.baggings should not be checked again. This should never be triggered.")
            self.baggings = 420

# Simple task that takes in an instance of the test dataclass above and prints it. Does no manipulation.
@task(log_prints=True)
def testTask(dc):
    print(dc)

# Simple flow that takes in a parameter, instantiates the test dataclass, and pass the dataclass to the task.
@flow(log_prints=True)
def testFlow(a: int):

    foo = testDC(a)
    testTask(foo)

if __name__ == "__main__":

    # Run the flow with a simple parameter
    testFlow(a=1)
    print(prefect.__version__)

12:32:41.624 | INFO    | prefect.engine - Created flow run 'russet-doberman' for flow 'testFlow'
12:32:41.775 | INFO    | Flow run 'russet-doberman' - post_init triggered
12:32:41.777 | INFO    | Flow run 'russet-doberman' - self.baggings is None triggered
12:32:41.811 | INFO    | Flow run 'russet-doberman' - Created task run 'testTask-0' for task 'testTask'
12:32:41.812 | INFO    | Flow run 'russet-doberman' - Executing 'testTask-0' immediately...
12:32:41.817 | INFO    | Flow run 'russet-doberman' - post_init triggered
12:32:41.818 | INFO    | Flow run 'russet-doberman' - self.baggings should not be checked again. This should never be triggered.
12:32:41.879 | INFO    | Task run 'testTask-0' - testDC(frodo=1, baggings=420)
12:32:41.923 | INFO    | Task run 'testTask-0' - Finished in state Completed()
12:32:41.969 | INFO    | Flow run 'russet-doberman' - Finished in state Completed('All states completed.')
2.8.6

Prefect 2.8.6

from prefect.utilities.annotations import quote

quote()