https://prefect.io logo
Title
n

Nic

10/28/2022, 9:21 AM
Hi - I have my task and flow in seperate files, but would like to use this
def cache_within_flow_run(context, parameters):
    return f"{context.flow_run_id}-{task_input_hash(context, parameters)}"
How would i pass the flow_run_id into the task this function and further into the task decorator?
@task(cache_key_fn=cache_within_flow_run)
def cached_task():
    print('running an expensive operation')
    return 42
1
m

Mason Menges

10/28/2022, 9:29 PM
Hey @Nic The context of the flow run isn't dependent on the file you're running your tasks in, i.e. both the flow and the task decorators are context aware as they're tied to the flow when it's running, if that makes sense. I don't think you would need to make any special changes in this circumstance.
n

Nic

11/01/2022, 8:48 AM
Hi @Mason Menges - When using it like described but calling task from flow and inserting the two codeblock in the task file i get get followring error
AttributeError: 'TaskRunContext' object has no attribute 'flow_run_id'
I was able to get around it with following code snippet
def cache_within_flow_run(context, parameters):
    cont_dict = context.task_run.dict()
    return f"{cont_dict['flow_run_id']}-{task_input_hash(context, parameters)}"
🙌 1