Paul Reyna
08/14/2023, 10:51 PMflow_name
and task_name
without having to explicitly define result_storage_key
for each task?from prefect import flow, task
from prefect.filesystems import LocalFileSystem, S3
@flow(result_storage=S3(bucket_path="my-bucket"))
def my_flow():
my_task()
@task(persist_result=True, result_storage_key="my_task.json")
def my_task():
...
my_flow() # The task's result will be persisted to '<s3://my-bucket/my_task.json>'
with Flow(
result=S3Result(bucket="MY_S3BUCKET", location=f"{flow_name}/{task_name}/{date}/{task_run_id}")
) as flow:
< flow code >
result_storage
, and nothing at the key levelflow_run , task_run
from prefect.runtime
in order to get those values, but I’m interested in the actual passing of those values into the S3 key