Ramtin
11/18/2021, 9:42 AMAnna Geller
from prefect import task, Flow
from prefect.engine.results import LocalResult
@task(result=LocalResult(location="sample_result.prefect"))
def hello_world():
return "Hello world"
with Flow("local-results") as flow:
hello_world()
state = flow.run()
hello_world = flow.get_tasks()[0]
print(hello_world.result.location)
Ramtin
11/18/2021, 9:54 AMRamtin
11/18/2021, 10:03 AM@task()
def hello_world():
infered_bucket = here i want the flow bucket
return "Hello world"
with Flow("local-results") as flow:
hello_world()
flow.result = S3Result(bucket="some_bucket")
Anna Geller
import boto3
from prefect.engine.results import S3Result
bucket = "some_bucket"
@task
def upload_object(file_name):
s3_client = boto3.client('s3')
s3_client.upload_file(file_name, bucket, f"path/on/s3/{file_name}")
with Flow("local-results") as flow:
upload_object()
flow.result = S3Result(bucket=bucket)