s3_bucket.py

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


@flow
async def aws_s3_bucket():
    aws_creds = await AwsCredentials.load("default")
    s3_bucket = S3Bucket(
        bucket_name="prefect-orion",
        aws_credentials=aws_creds,
        basepath="docs",
    )
    files = await s3_bucket.put_directory(local_path="docs")
    print(files)


if __name__ == "__main__":
    asyncio.run(aws_s3_bucket())