Brian Newman
09/19/2023, 2:55 AMChristopher Boyd
09/19/2023, 1:36 PMBrian Newman
09/19/2023, 4:32 PMfrom prefect.filesystems import Azure
def generate_task_name():
flow_name = flow_run.flow_name
task_name = task_run.task_name
return f"{flow_name}::TASK::{task_name}"
def cache_with_storage(context, parameters):
storage_block_value = context.result_factory.storage_block
return f"{storage_block_value}-{task_input_hash(context, parameters)}"
@task(
cache_key_fn=cache_with_storage,
result_storage=Azure.load("prefect-cache"),
cache_expiration=timedelta(days=7),
log_prints=True,
task_run_name=generate_task_name,
)
async def fetch_data() -> str:
Christopher Boyd
09/19/2023, 4:37 PM