from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.s3 import s3_upload

@flow
def example_s3_upload_flow():
    aws_credentials = AwsCredentials(
        aws_access_key_id="acccess_key_id",
        aws_secret_access_key="secret_access_key"
    )
    with open("data.csv", "rb") as file:
        key = s3_upload(
            bucket="bucket",
            key="data.csv",
            data=file.read(),
            aws_credentials=aws_credentials,
        )

example_s3_upload_flow()

import asyncio
from prefect import flow
from prefect_aws import AwsCredentials
from prefect_aws.s3 import S3Bucket


@flow
async def aws_s3_bucket_roundtrip():
    # create an AwsCredentials block here or through UI
    aws_creds = AwsCredentials(
        aws_access_key_id="AWS_ACCESS_KEY_ID",
        aws_secret_access_key="AWS_SECRET_ACCESS_KEY"
    )

    s3_bucket = S3Bucket(
        bucket_name="bucket",  # must exist
        aws_credentials=aws_creds,
        basepath="subfolder",
    )

    key = await s3_bucket.write_path("data.csv", content=b"hello")

    return await s3_bucket.read_path(key)

asyncio.run(aws_s3_bucket_roundtrip())

import asyncio
from prefect import flow
from prefect_aws import MinIOCredentials
from prefect_aws.s3 import S3Bucket

@flow
async def minio_s3_bucket_roundtrip():

    minio_creds = MinIOCredentials.load("MY_BLOCK_NAME")

    s3_bucket = S3Bucket(
        bucket_name="bucket",  # must exist
        minio_credentials=minio_creds,
        endpoint_url="http://localhost:9000"
    )

    path_to_file = await s3_bucket.write_path("/data.csv", content=b"hello")
    return await s3_bucket.read_path(path_to_file)

asyncio.run(minio_s3_bucket_roundtrip())