https://prefect.io logo
p

Paul Reyna

08/14/2023, 10:51 PM
hello! is it possible to have the S3 key for storing results include stuff like
flow_name
and
task_name
without having to explicitly define
result_storage_key
for each task?
I’m seeing examples like this
Copy code
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>'
in Prefect 1, we were able to do something like this, where we only had to define it at the flow definition, rather than explicitly defining the S3 key for each task
Copy code
with Flow(
    result=S3Result(bucket="MY_S3BUCKET", location=f"{flow_name}/{task_name}/{date}/{task_run_id}")
    ) as flow:
    < flow code >
I was wondering if there’s some sort of equivalent in Prefect 2, but it seems like the flow decorator only allows us to pass in the S3 bucket via
result_storage
, and nothing at the key level
also, I see from https://docs.prefect.io/2.11.3/concepts/tasks/ that we import
flow_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