Hello all ! I'm looking for a simple way to modify a PrefectResult, in order to re-run a task with t...

Didier Marin

01/12/2022, 5:32 PM

Hello all ! I'm looking for a simple way to modify a PrefectResult, in order to re-run a task with the correct input without re-running the whole flow. Is there a way to do it in Python, or otherwise with a mutation in GraphQL? I explored the schema but could not pinpoint where to find the result given its "location" id 🤔

Kevin Kho

01/12/2022, 5:45 PM

You need to do take advantage of something like caching so that the task output is used for the next flow rurn

Didier Marin

01/12/2022, 6:38 PM

I'll check it out, thank you Kevin!

Didier Marin

01/13/2022, 2:43 PM

I ended up doing something like this:

Copy code

from prefect import Client
from prefect.engine.results import PrefectResult
from flows.serializers import Base64PickleSerializer

# The following info can be found by navigating to the task run in the UI
task_run_id = ... # Your task run's id, e.g. "1cb9f9d4-9cdc-4606-b6d4-7a62cb82c4d5"
location = ... # Location of the task run's result, e.g. "gASVCQAAAAAAAABdlIwCZXWUYS4="

new_value = ... # New value to set up

c = Client()
# Get the original state
s = c.get_task_run_state()
# Override the state's result
r = PrefectResult(location=location, serializer=Base64PickleSerializer())
r = r.write(new_value)
s.result = r
# Update the state of the task_run
new_s = c.set_task_run_state(task_run_id, s)

12 Views

Open in Slack

Previous Next

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.