Repository: PrefectHQ/prefect
File: src/integrations/prefect-aws/prefect_aws/s3.py (Python)

Code before:
from prefect.blocks.abstract import CredentialsBlock, ObjectStorageBlock
from prefect.filesystems import WritableDeploymentStorage, WritableFileSystem
from prefect.logging import get_run_logger
from prefect.utilities.asyncutils import run_sync_in_worker_thread
from prefect.utilities.filesystem import filter_files
from prefect.utilities.pydantic import lookup_type
from prefect_aws import AwsCredentials, MinIOCredentials
from prefect_aws.client_parameters import AwsClientParameters


def get_s3_client(
    credentials: Optional[dict[str, Any]] = None,
    client_parameters: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """
    Get a boto3 S3 client with the given credentials and client parameters.

    Args:
        credentials: A dictionary of credentials to use for authentication with AWS.
        client_parameters: A dictionary of parameters to use for the boto3 client
            initialization.

    Returns:
        A boto3 S3 client.
    """
    credentials = credentials or {}
    client_parameters = client_parameters or {}

    # Handle aliasing MinIO settings to AWS credentials
    credentials.setdefault(
        "aws_access_key_id", credentials.pop("minio_root_user", None)
    )
    credentials.setdefault(
        "aws_secret_access_key", credentials.pop("minio_root_password", None)
    )

    # Use the explicitly provided client parameters if not already defined in the credentials block
    credentials.setdefault(
        "aws_client_parameters",
        client_parameters and AwsClientParameters.model_validate(client_parameters),
    )

    return AwsCredentials.model_validate(credentials).get_s3_client()


@task
async def adownload_from_bucket(
    bucket: str,
    key: str,
    aws_credentials: AwsCredentials,
    aws_client_parameters: AwsClientParameters = AwsClientParameters(),

Review comment (security):
nit: 45-50 would mutate the caller's data, shouldn't actually matter but it's safer to make a copy i think
```suggestion
    credentials = dict(credentials or {})
```

Code after:
from prefect.blocks.abstract import CredentialsBlock, ObjectStorageBlock
from prefect.filesystems import WritableDeploymentStorage, WritableFileSystem
from prefect.logging import get_run_logger
from prefect.utilities.asyncutils import run_sync_in_worker_thread
from prefect.utilities.filesystem import filter_files
from prefect.utilities.pydantic import lookup_type
from prefect_aws import AwsCredentials, MinIOCredentials
from prefect_aws.client_parameters import AwsClientParameters


def get_s3_client(
    credentials: Optional[dict[str, Any]] = None,
    client_parameters: Optional[dict[str, Any]] = None,
) -> dict[str, Any]:
    """
    Get a boto3 S3 client with the given credentials and client parameters.

    Args:
        credentials: A dictionary of credentials to use for authentication with AWS.
        client_parameters: A dictionary of parameters to use for the boto3 client
            initialization.

    Returns:
        A boto3 S3 client.
    """
    credentials = dict(credentials or {})
    client_parameters = client_parameters or {}

    # Handle aliasing MinIO settings to AWS credentials
    credentials.setdefault(
        "aws_access_key_id", credentials.pop("minio_root_user", None)
    )
    credentials.setdefault(
        "aws_secret_access_key", credentials.pop("minio_root_password", None)
    )

    # Use the explicitly provided client parameters if not already defined in the credentials block
    credentials.setdefault(
        "aws_client_parameters",
        client_parameters and AwsClientParameters.model_validate(client_parameters),
    )

    return AwsCredentials.model_validate(credentials).get_s3_client()


@task
async def adownload_from_bucket(
    bucket: str,
    key: str,
    aws_credentials: AwsCredentials,
    aws_client_parameters: AwsClientParameters = AwsClientParameters(),