The subclassed tasks look like this:

class WriteMessageToFileWithDelayTask(Task):

def run(self, message: str, delay: int, filepath: str) -> bool:

return write_message_to_file_with_delay.run(message, delay, filepath)

Could the problem be the way I am calling the decorated task in the subclassed task?!

(Update: I tried putting the task code directly into the run method instead of calling the decorated task but that didn’t seem to help. 😞)

I’ll take a look at the link, thanks! (It looks promising!)

Sorry for the code dump but maybe this will be useful - this shows the way I define the two flows. Where I register/run the flow programmatically, I switch between “get_file_io_flow” and “get_serializable_file_io_flow” and see the problem only in the first version. NOTE - functions are defined module

file_io_flows

- I import this in yet a third file, where I set up the storage and register the flow as described above.

from example_dags.file_io_tasks import (

ReadMessageFromFileWithDelayTask,

WriteMessageToFileWithDelayTask,

read_message_from_file_with_delay,

write_message_to_file_with_delay

)

def get_file_io_flow(name: str = _DEFAULT_FILE_IO_FLOW_NAME) -> Tuple[Flow, Task]:

"""Imperative style Flow demonstrating use of subclassed Tasks."""

read_delay = Parameter('read_delay', default=0)

write_delay = Parameter('write_delay', default=0)

message = Parameter('message', default='HELLO')

filepath = Parameter('filepath', default='./test_file.txt')

read_message = ReadMessageFromFileWithDelayTask()

write_message = WriteMessageToFileWithDelayTask()

flow = Flow(name, tasks = [read_message, write_message])

flow.set_dependencies(task=read_message,

upstream_tasks=[write_message, read_delay, filepath],

keyword_tasks={'delay': read_delay,

'filepath': filepath})

flow.set_dependencies(task=write_message,

upstream_tasks=[write_delay, filepath, message],

keyword_tasks={'delay': write_delay,

'filepath': filepath,

'message': message})

return flow, read_message

def get_serializable_file_io_flow(name: str = _DEFAULT_FILE_IO_FLOW_NAME) -> Tuple[Flow, Task]:

"""Functional Flow that uses decorated tasks.

Using this for Docker storage run since the imperative get_file_io_flow resulted in an error

that something was not serializable.

"""

with Flow(name) as flow:

read_delay = Parameter('read_delay', default=0)

write_delay = Parameter('write_delay', default=0)

message = Parameter('message', default='HELLO')

filepath = Parameter('filepath', default='./test_file.txt')

write_message = write_message_to_file_with_delay(message, write_delay, filepath)

read_message = read_message_from_file_with_delay(read_delay, filepath)

flow.set_dependencies(task=read_message, upstream_tasks=[write_message])

return flow, read_message

And here is an example of how I define the subclassed vs decorated flow in `file_io_tasks`:

@task

def write_message_to_file_with_delay(message: str,

delay: int,

filepath: str) ->bool:

"""Writes message to file after delay."""

sleep(delay)

with open(filepath, 'w', encoding='utf-8') as stream:

stream.write(message)

logger = prefect.context.get("logger")

<http://logger.info|logger.info>(f'*~*~* wrote message to file after {delay} second(s) *~*~*')

return True

class WriteMessageToFileWithDelayTask(Task):

def run(self, message: str, delay: int, filepath: str) -> bool:

sleep(delay)

with open(filepath, 'w', encoding='utf-8') as stream:

stream.write(message)

logger = prefect.context.get("logger")

<http://logger.info|logger.info>(f'*~*~* wrote message to file after {delay} second(s) *~*~*')

return True

I think the

@task

is serialized along with the Flow but looks like the custom Task class is not

Did you try running the ones defined with a decorator and seeing if it works? I wouldn’t be surprised if it failed at run time.

For posterity: I think my first interpretation wasn’t exactly right: I stepped through the serialization in the debugger and noticed that the type of the serialized decorated task is

'prefect.tasks.core.function.FunctionTask'

, but the type of the serialized subclassed task is

'example_dags.file_io_tasks.WriteMessageToFileWithDelayTask'

. The first is defined in the executor but the second is not. To deserialize, the type must be defined! So I think it might be more correct to say that both are serialized correctly, but the “shape of the vessel” that they get deserialized to is available in the decorated case but not in the subclassed case.