Hi, I found a bug with the flow idempotency key generation. When you use an object created with

typing.NewType()

for a task type hint, the idempotency key becomes non-deterministic. From what I can tell, it’s because

typing.NewType()

actually returns a function, and calling

repr

or

str

on a function includes the memory address in the result. Here is a simple flow that reproduces the issue:

import prefect
import typing


MyNewType = typing.NewType("MyNewType", str)


@prefect.task(name="print_something")
def print_something(input_str: MyNewType):
    print(input_str)


with prefect.Flow("my_flow") as flow:
    print_something("test_str")

# This is different every time script is run
print(flow.serialized_hash())

# Output includes memory address, e.g.
# "<function NewType.<locals>.new_type at 0x7fa6c71c34c0>"
print(list(flow.tasks)[0].inputs()["input_str"]["type"]

As for fixing this, I don’t know enough about the serialization process to know where to start. It can be mitigated by wrapping the type hint in

typing.Optional[]

. I’m running prefect version 0.14.22 on python 3.9.1

Hey @Paul Schale thanks for the report. I'll open an issue on Github for this.