Hi! Quick question: I'm using the GCSResultHandler with a task and I'd like to write the URI safe value of the result handler to a database. Is it possible during flow runtime to access a ResultHandler to persist the safe value?

Hi Cab, great question! Are you running purely in Core or with Cloud as well?

We're running with core for local development and cloud in staging/prod environments so we'll need to be able to access in both environments

Okay, gotcha. Asking the team to see what advice we can give-- will let you know as soon as we've got an answer.

👍 awesome thanks!

Back with a couple different suggestions: • if you're working with a Task class definition, you can reference your result handler during the

run

method if you've set

self.result_handler

(which it sounds like you have) • you could also add a state handler to your task and access the result info you're looking for by accessing

new_state._result

OK, sounds good. I'm thinking if I need to use the safe value of the result, I'll need to us the second one since the result handler won't produce the safe value until after the task is completed. I'll look into using a state handler

@Zachary Hughes Any idea why a state handler function would be running twice? I wrote a function that I am using as a state handler for a task, and I have a print statement inside of it, and it is getting called twice.

Not off the top of my head! Give me just a sec and I can take a look.

hmm, that depends on what is considered a state change. I am only calling this task once, so I would think that there is an inital state, and then the task gets called and now there is a new state. I don't know what would be causing the second state change

Ah, okay. When we talk about states, we talk about Prefect states. So the state progression there is Pending -> Running -> Success. So even though you're only running it once, it goes through a minimum of two state changes.

When I print out the new_state object in the result handler, they are very similar, like they have the same state. But I think I found out the source of the issue. The task that has this state handler on it also uses caching. the task decorator looks something like this:

@task(
    checkpoint=True,
    result_handler=GCSResultHandler(
        bucket="my-bucket", credentials_secret=GCS_CREDS_SECRET_NAME
    ),
    state_handlers=[my_state_handler_function],
    cache_for=datetime.timedelta(days=DEFAULT_TASK_CACHE_DURATION),
    cache_validator=cache_validators.all_inputs,
    cache_key=CACHE_KEYS.get("run_function_foo"),
)
def run_function_foo(arg1: str, arg2: str) -> str:

When I take out those last three caching arguments to the task decorator then the state handler only runs once

When you're working with cached tasks, that actually adds another Prefect state to the mix:

Cached

. So in that case, you have another state transition:

Success

->

Cached

. But if you're just looking to trigger this logic on

Success

, you should be able to check the

new_state

argument in your state handler and branch based on that.