Didier Marin

    Didier Marin

    8 months ago
    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

    Kevin Kho

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

    Didier Marin

    8 months ago
    I'll check it out, thank you Kevin!
    I ended up doing something like this:
    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)