Henri
09/26/2023, 11:00 PMfrom prefect.filesystems import S3
import os
def s3_storage() -> None:
block = S3(
bucket_path="prefect-portal-us-east-1-qa",
aws_access_key_id=os.getenv("QA_AWS_ACCESS_KEY_ID"),
aws_secret_access_key=os.getenv("QA_AWS_SECRET_ACCESS_KEY"),
).save("ecs-s3-qa", overwrite=True)
if __name__ == "__main__":
s3_storage()
An example
from prefect_aws import AwsCredentials
from prefect_aws.ecs import ECSTask
from prefect import flow, task
from prefect.deployments.deployments import Deployment
from prefect.filesystems import S3
[...]
@flow(log_prints=True)
def cool_numbers(nums=[1, 2, 3, 5, 8, 13]): # essentially map_flow
print_nums(nums)
squared_nums = square_num.map(nums)
print_nums(squared_nums)
if __name__ == "__main__":
aws_credentials_block = AwsCredentials.load("aws-credentials-qa") # loads my aws cred
ecs_task_block = ECSTask.load("ecs-cluster-agent-qa") # refers to the aws ecs cluster using
s3_block = S3.load("ecs-s3-qa") # doesn't really work
deployment = Deployment.build_from_flow(
flow=cool_numbers,
name="ecs-cool-numbers",
work_queue_name="default",
infrastructure=ecs_task_block,
path="/",
apply=True,
tags=["ecs-testing"],
)
The tl;dr is that prefect can't seem to upload to s3 and I can't execute the flow as its looking in s3.Henri
09/26/2023, 11:02 PM