@Marvin why might the return value of my async task be None only when using

.submit

on my task?

It appears you're encountering an issue where the return value of your asynchronous task is

None

when using

.submit

method. From the GitHub issues I found in the Prefect repository, there doesn't seem to be an explicitly matching case. However, a similar issue arises when using

.submit

with the SequentialTaskRunner, and the tasks do not respect the defined sequence, potentially due to exceptions not properly causing subsequent tasks to cease execution. The proper use of

wait_for

can rectify this issue, indicating that managing task dependencies might play a role in the unexpected

None

return value you're seeing (#12804). Other related issues involve tasks not finishing as expected when errors occur in mapped tasks (#12028), and exceptions raised when running flows in parallel (#9296). Without a direct match, I would recommend doing the following: 1. Ensure that your task's return is not being overridden or manipulated unexpectedly. 2. Examine the task logs to see if there is an exception being raised within the task itself that could result in a

None

return. 3. Check if the dependencies between tasks are correctly managed using

wait_for

to ensure tasks are not running out of order. If you're still encountering the problem, you might consider raising a new GitHub issue with detailed information about your specific use case so that the Prefect community can help identify the problem.

hi @Matthew Bell - that sounds odd. do you have a minimal example?

Hi Nate I just posted and solved my problem in this thread. Is this the intended syntax required or is this a bug? Seems really unintuitive and isn't documented properly as far as I can tell

sorry, intended required syntax to do what?

To get both the state and result of an async task

i think

return_state

may be what you want? youd need to

persist_result

if you want the result via the state though

In [1]: from prefect import task, flow

In [2]: @task(persist_result=True)
   ...: async def t():
   ...:     return 42
   ...:

In [3]: @flow
   ...: async def f():
   ...:     s = await t.submit(return_state=True)
   ...:     assert s.is_completed()
   ...:     return await s.result().get()
   ...:

In [4]: assert await f() == 42
19:39:09.234 | INFO    | prefect.engine - Created flow run 'modest-aardwark' for flow 'f'
19:39:09.544 | INFO    | Flow run 'modest-aardwark' - Created task run 't-0' for task 't'
19:39:09.877 | INFO    | Flow run 'modest-aardwark' - Submitted task run 't-0' for execution.
19:39:10.204 | INFO    | Task run 't-0' - Finished in state Completed()
19:39:10.348 | INFO    | Flow run 'modest-aardwark' - Finished in state Completed()

Ahh state is retrieving the persisted result. I did not realize this

👍