https://prefect.io logo
Join Slack
Powered by
Hi <@ULVA73B9P> In my tests, I have a flow that re...
# ask-marvin
v
Hi @Marvin In my tests, I have a flow that returns multiple results :
Copy code
@flow(name="Filter flow")
def filter_flow():
    li = some_list()
    funky_li = unstable_task.map(li)
    filtered_li = filter_results(allow_failure(funky_li))
    added_li = add_one.map(filtered_li)
    return (
        li,
        funky_li,
        filtered_li,
        added_li
    )
I then test this flow with :
Copy code
def test_filter_failed_tasks():
    state = filter_flow(return_state=True)
    assert state.is_completed()
    (
        li,
        funky_li,
        filtered_li,
        added_li,
    ) = state.result()
    assert li == [0, 1, 2, 3, 4]
    assert len(funky_li) == 5
    assert filtered_li == [0, 4]
    assert added_li.result() == [1, 5]
This works with Prefect up to 3.6.16 and fails from 3.6.17+ with :
Copy code
>       assert added_li.result() == [1, 5]
[...]
E                   AttributeError: 'State' object has no attribute '_final_state'
m
❌ Error: unhandled errors in a TaskGroup (1 sub-exception)
Sorry, I encountered an error while processing your request. Please try again.
v
Hi @Marvin I have a flow that returns multiple results :
Copy code
@flow(name="Filter flow")
def filter_flow():
    li = some_list()
    funky_li = unstable_task.map(li)
    filtered_li = filter_results(allow_failure(funky_li))
    added_li = add_one.map(filtered_li)
    return (
        li,
        funky_li,
        filtered_li,
        added_li
    )
I then test this flow with :
Copy code
def test_filter_failed_tasks():
    state = filter_flow(return_state=True)
    assert state.is_completed()
    (
        li,
        funky_li,
        filtered_li,
        added_li,
    ) = state.result()
    assert li == [0, 1, 2, 3, 4]
    assert len(funky_li) == 5
    assert filtered_li == [0, 4]
    assert added_li.result() == [1, 5]
This works with Prefect up to 3.6.16 and fails from 3.6.17+ with :
Copy code
>       assert added_li.result() == [1, 5]
[...]
E                   AttributeError: 'State' object has no attribute '_final_state'
m
✅ thought for 318.8 seconds
Thanks for the clear repro — this is very helpful. You’ve run into a regression introduced in 3.6.17 that affects mapped results returned from a flow when you access them via a flow State. Specifically, when a flow returns a PrefectFutureList and you call 
state.result()
(because you ran the flow with 
return_state=True
), the container can end up as a PrefectFutureList whose elements are States instead of PrefectFutures. Then calling 
added_li.result()
tries to treat those elements like PrefectFutures and triggers:
Copy code
AttributeError: 'State' object has no attribute '_final_state'
Workarounds you can use right now: - Simplest (and recommended for unit tests): don’t use 
return_state=True
when you just want to assert on the flow’s return value. Let the flow resolve its return value directly:
Copy code
def test_filter_failed_tasks():
      li, funky_li, filtered_li, added_li = filter_flow()
      assert li == [0, 1, 2, 3, 4]
      assert len(funky_li) == 5
      assert filtered_li == [0, 4]
      assert added_li == [1, 5]
- If you need to assert on the flow state as well, you can still do so and then compare values without calling 
.result()
on the mapped container:
Copy code
state = filter_flow(return_state=True)
  assert state.is_completed()
  li, funky_li, filtered_li, added_li = state.result()

  # On affected versions, elements of `added_li` are States
  assert [s.result() for s in added_li] == [1, 5]
- Or resolve mapped futures inside the flow before returning:
Copy code
@flow(name="Filter flow")
  def filter_flow():
      li = some_list()
      funky_li = unstable_task.map(li)
      filtered_li = filter_results(allow_failure(funky_li))
      added_li = add_one.map(filtered_li)
      return li, funky_li.result(), filtered_li, added_li.result()
Next steps: - If you can, upgrade to the latest 3.6.x — this regression has been fixed in a subsequent 3.6 release. If you’re pinned to 3.6.17–3.6.23 for now, use one of the workarounds above or temporarily pin to 3.6.16. - Release notes: Prefect releases - Docs for context: - States and results: States and results - Tasks and mapping: Tasks (mapping) If you’re still seeing this on the latest 3.6.x, let me know the exact Prefect version you tested and I can open an issue with your minimal repro.
Open in Slack
PreviousNext
Powered by