From c584ed721234422d8540d10229ef6c301a89f0d6 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Tue, 19 Dec 2023 17:08:00 +0000
Subject: [PATCH 01/30] move s3_client up

---
 prefect_aws/s3.py | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 643d78ac..8dc29034 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -551,6 +551,8 @@ async def put_directory(
 
             included_files = filter_files(local_path, ignore_patterns)
 
+        s3_client = self._get_s3_client()
+
         uploaded_file_count = 0
         for local_file_path in Path(local_path).expanduser().rglob("*"):
             if (
@@ -566,7 +568,9 @@ async def put_directory(
                     local_file_content = local_file.read()
 
                 await self.write_path(
-                    remote_file_path.as_posix(), content=local_file_content
+                    s3_client,
+                    remote_file_path.as_posix(),
+                    content=local_file_content,
                 )
                 uploaded_file_count += 1
 
@@ -620,7 +624,9 @@ def _read_sync(self, key: str) -> bytes:
             return output
 
     @sync_compatible
-    async def write_path(self, path: str, content: bytes) -> str:
+    async def write_path(
+        self, s3_client: boto3.client, path: str, content: bytes
+    ) -> str:
         """
         Writes to an S3 bucket.
 
@@ -654,11 +660,11 @@ async def write_path(self, path: str, content: bytes) -> str:
 
         path = self._resolve_path(path)
 
-        await run_sync_in_worker_thread(self._write_sync, path, content)
+        await run_sync_in_worker_thread(self._write_sync, s3_client, path, content)
 
         return path
 
-    def _write_sync(self, key: str, data: bytes) -> None:
+    def _write_sync(self, s3_client: boto3.client, key: str, data: bytes) -> None:
         """
         Called by write_path(). Creates an S3 client and uploads a file
         object.

From 70ba4b7cb4f291f353a5cdebc317c8d13b691086 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Tue, 19 Dec 2023 17:27:57 +0000
Subject: [PATCH 02/30] get s3 client

---
 prefect_aws/s3.py | 2 --
 1 file changed, 2 deletions(-)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 8dc29034..8e15bf73 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -670,8 +670,6 @@ def _write_sync(self, s3_client: boto3.client, key: str, data: bytes) -> None:
         object.
         """
 
-        s3_client = self._get_s3_client()
-
         with io.BytesIO(data) as stream:
             s3_client.upload_fileobj(Fileobj=stream, Bucket=self.bucket_name, Key=key)
 

From 2389cbe714b72c7336dc9c3312a366a543c9654e Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Tue, 19 Dec 2023 17:32:20 +0000
Subject: [PATCH 03/30] add print for testing

---
 prefect_aws/s3.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 8e15bf73..337fcebf 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -573,6 +573,7 @@ async def put_directory(
                     content=local_file_content,
                 )
                 uploaded_file_count += 1
+                print(uploaded_file_count)  # TODO remove print
 
         return uploaded_file_count
 

From 4a7440128496848e51c40023f67da82a379c27b3 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Tue, 19 Dec 2023 18:27:22 +0000
Subject: [PATCH 04/30] revert changes

---
 prefect_aws/s3.py | 16 ++++++----------
 1 file changed, 6 insertions(+), 10 deletions(-)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 337fcebf..65bb026f 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -551,8 +551,6 @@ async def put_directory(
 
             included_files = filter_files(local_path, ignore_patterns)
 
-        s3_client = self._get_s3_client()
-
         uploaded_file_count = 0
         for local_file_path in Path(local_path).expanduser().rglob("*"):
             if (
@@ -568,9 +566,7 @@ async def put_directory(
                     local_file_content = local_file.read()
 
                 await self.write_path(
-                    s3_client,
-                    remote_file_path.as_posix(),
-                    content=local_file_content,
+                    remote_file_path.as_posix(), content=local_file_content
                 )
                 uploaded_file_count += 1
                 print(uploaded_file_count)  # TODO remove print
@@ -625,9 +621,7 @@ def _read_sync(self, key: str) -> bytes:
             return output
 
     @sync_compatible
-    async def write_path(
-        self, s3_client: boto3.client, path: str, content: bytes
-    ) -> str:
+    async def write_path(self, path: str, content: bytes) -> str:
         """
         Writes to an S3 bucket.
 
@@ -661,16 +655,18 @@ async def write_path(
 
         path = self._resolve_path(path)
 
-        await run_sync_in_worker_thread(self._write_sync, s3_client, path, content)
+        await run_sync_in_worker_thread(self._write_sync, path, content)
 
         return path
 
-    def _write_sync(self, s3_client: boto3.client, key: str, data: bytes) -> None:
+    def _write_sync(self, key: str, data: bytes) -> None:
         """
         Called by write_path(). Creates an S3 client and uploads a file
         object.
         """
 
+        s3_client = self._get_s3_client()
+
         with io.BytesIO(data) as stream:
             s3_client.upload_fileobj(Fileobj=stream, Bucket=self.bucket_name, Key=key)
 

From 362c97f3d043032ab72f225568f338969ef13b48 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Tue, 19 Dec 2023 18:46:47 +0000
Subject: [PATCH 05/30] update client

---
 prefect_aws/credentials.py | 9 ++++++++-
 1 file changed, 8 insertions(+), 1 deletion(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 64f49efe..af9e7f75 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -75,6 +75,8 @@ class AwsCredentials(CredentialsBlock):
         title="AWS Client Parameters",
     )
 
+    _s3_client: Optional[S3Client] = None
+
     def get_boto3_session(self) -> boto3.Session:
         """
         Returns an authenticated boto3 session that can be used to create clients
@@ -132,7 +134,12 @@ def get_s3_client(self) -> S3Client:
         Returns:
             An authenticated S3 client.
         """
-        return self.get_client(client_type=ClientType.S3)
+        if self._s3_client is not None:
+            return self._s3_client
+
+        self._s3_client = self.get_client(client_type=ClientType.S3)
+
+        return self._s3_client
 
     def get_secrets_manager_client(self) -> SecretsManagerClient:
         """

From 5feb2000c95852dba7152cbab9554f8ae6f6e0a3 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 13:07:17 +0100
Subject: [PATCH 06/30] remove print

---
 prefect_aws/s3.py | 1 -
 1 file changed, 1 deletion(-)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 65bb026f..643d78ac 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -569,7 +569,6 @@ async def put_directory(
                     remote_file_path.as_posix(), content=local_file_content
                 )
                 uploaded_file_count += 1
-                print(uploaded_file_count)  # TODO remove print
 
         return uploaded_file_count
 

From 43fad43a22ae22fcebda79d93bbab39f5d288a51 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 15:10:59 +0100
Subject: [PATCH 07/30] add @lru_cache

---
 prefect_aws/credentials.py | 34 ++++++++++++++++++++--------------
 1 file changed, 20 insertions(+), 14 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index af9e7f75..dfc530f1 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -1,6 +1,7 @@
 """Module handling AWS credentials"""
 
 from enum import Enum
+from functools import lru_cache
 from typing import Any, Optional, Union
 
 import boto3
@@ -24,6 +25,18 @@ class ClientType(Enum):
     SECRETS_MANAGER = "secretsmanager"
 
 
+@lru_cache
+def get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
+    if isinstance(client_type, ClientType):
+        client_type = client_type.value
+
+    client = ctx.get_boto3_session().client(
+        service_name=client_type,
+        **ctx.aws_client_parameters.get_params_override(),
+    )
+    return client
+
+
 class AwsCredentials(CredentialsBlock):
     """
     Block used to manage authentication with AWS. AWS authentication is
@@ -75,7 +88,11 @@ class AwsCredentials(CredentialsBlock):
         title="AWS Client Parameters",
     )
 
-    _s3_client: Optional[S3Client] = None
+    class Config:
+        arbitrary_types_allowed = True
+
+    def __hash__(self):
+        return id(self)
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -119,13 +136,7 @@ def get_client(self, client_type: Union[str, ClientType]) -> Any:
         Raises:
             ValueError: if the client is not supported.
         """
-        if isinstance(client_type, ClientType):
-            client_type = client_type.value
-
-        client = self.get_boto3_session().client(
-            service_name=client_type, **self.aws_client_parameters.get_params_override()
-        )
-        return client
+        return get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
         """
@@ -134,12 +145,7 @@ def get_s3_client(self) -> S3Client:
         Returns:
             An authenticated S3 client.
         """
-        if self._s3_client is not None:
-            return self._s3_client
-
-        self._s3_client = self.get_client(client_type=ClientType.S3)
-
-        return self._s3_client
+        return self.get_client(client_type=ClientType.S3)
 
     def get_secrets_manager_client(self) -> SecretsManagerClient:
         """

From 1df6cb3f522da7db7cd77640949c0c70faad4a7b Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 16:02:49 +0100
Subject: [PATCH 08/30] Add docs

---
 prefect_aws/credentials.py | 36 ++++++++++++++-----
 tests/test_credentials.py  | 71 ++++++++++++++++++++++++++++++++++++--
 2 files changed, 96 insertions(+), 11 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index dfc530f1..b285a340 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -19,6 +19,8 @@
 
 
 class ClientType(Enum):
+    """Boto3 Client Types"""
+
     S3 = "s3"
     ECS = "ecs"
     BATCH = "batch"
@@ -26,7 +28,19 @@ class ClientType(Enum):
 
 
 @lru_cache
-def get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
+def _get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
+    """
+    Helper method to cache and  dynamically get a client type.
+
+    Args:
+        client_type: The client's service name.
+
+    Returns:
+        An authenticated client.
+
+    Raises:
+        ValueError: if the client is not supported.
+    """
     if isinstance(client_type, ClientType):
         client_type = client_type.value
 
@@ -89,6 +103,8 @@ class AwsCredentials(CredentialsBlock):
     )
 
     class Config:
+        """pydantic config"""
+
         arbitrary_types_allowed = True
 
     def __hash__(self):
@@ -136,7 +152,7 @@ def get_client(self, client_type: Union[str, ClientType]) -> Any:
         Raises:
             ValueError: if the client is not supported.
         """
-        return get_client_cached(ctx=self, client_type=client_type)
+        return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
         """
@@ -199,6 +215,14 @@ class MinIOCredentials(CredentialsBlock):
         description="Extra parameters to initialize the Client.",
     )
 
+    class Config:
+        """pydantic config"""
+
+        arbitrary_types_allowed = True
+
+    def __hash__(self):
+        return id(self)
+
     def get_boto3_session(self) -> boto3.Session:
         """
         Returns an authenticated boto3 session that can be used to create clients
@@ -244,13 +268,7 @@ def get_client(self, client_type: Union[str, ClientType]) -> Any:
         Raises:
             ValueError: if the client is not supported.
         """
-        if isinstance(client_type, ClientType):
-            client_type = client_type.value
-
-        client = self.get_boto3_session().client(
-            service_name=client_type, **self.aws_client_parameters.get_params_override()
-        )
-        return client
+        return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
         """
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 6e0a1ff8..a69d3eb9 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -3,7 +3,12 @@
 from botocore.client import BaseClient
 from moto import mock_s3
 
-from prefect_aws.credentials import AwsCredentials, ClientType, MinIOCredentials
+from prefect_aws.credentials import (
+    AwsCredentials,
+    ClientType,
+    MinIOCredentials,
+    _get_client_cached,
+)
 
 
 def test_aws_credentials_get_boto3_session():
@@ -41,6 +46,68 @@ def test_minio_credentials_get_boto3_session():
     ],
 )
 @pytest.mark.parametrize("client_type", ["s3", ClientType.S3])
-def test_credentials_get_client(credentials, client_type):
+def test_credentials_get_client_s3(credentials, client_type):
+    with mock_s3():
+        assert isinstance(credentials.get_client(client_type), BaseClient)
+
+
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
+@pytest.mark.parametrize("client_type", ["ecs", ClientType.ECS])
+def test_credentials_get_client_ecs(credentials, client_type):
+    with mock_s3():
+        assert isinstance(credentials.get_client(client_type), BaseClient)
+
+
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
+@pytest.mark.parametrize("client_type", ["batch", ClientType.BATCH])
+def test_credentials_get_client_batch(credentials, client_type):
     with mock_s3():
         assert isinstance(credentials.get_client(client_type), BaseClient)
+
+
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
+@pytest.mark.parametrize("client_type", ["s3", ClientType.S3])
+def test_credentials_get_client_cached_s3(credentials, client_type):
+    with mock_s3():
+        assert isinstance(_get_client_cached(credentials, client_type), BaseClient)
+
+
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
+@pytest.mark.parametrize("client_type", ["ecs", ClientType.ECS])
+def test_credentials_get_client_cached_ecs(credentials, client_type):
+    with mock_s3():
+        client = _get_client_cached(credentials, client_type)
+
+        assert isinstance(client, BaseClient)

From e23c67b977ca711d6b90f1c1c10b8d169ac00d59 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 16:07:57 +0100
Subject: [PATCH 09/30] update docs

---
 prefect_aws/credentials.py |   2 +-
 tests/test_credentials.py  | 340 ++++++++++++++++++++++++++-----------
 2 files changed, 239 insertions(+), 103 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index b285a340..0ebfe787 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -19,7 +19,7 @@
 
 
 class ClientType(Enum):
-    """Boto3 Client Types"""
+    """The supported boto3 clients."""
 
     S3 = "s3"
     ECS = "ecs"
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index a69d3eb9..64f49efe 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -1,113 +1,249 @@
-import pytest
-from boto3.session import Session
-from botocore.client import BaseClient
-from moto import mock_s3
+"""Module handling AWS credentials"""
 
-from prefect_aws.credentials import (
-    AwsCredentials,
-    ClientType,
-    MinIOCredentials,
-    _get_client_cached,
-)
+from enum import Enum
+from typing import Any, Optional, Union
 
+import boto3
+from mypy_boto3_s3 import S3Client
+from mypy_boto3_secretsmanager import SecretsManagerClient
+from prefect.blocks.abstract import CredentialsBlock
+from pydantic import VERSION as PYDANTIC_VERSION
 
-def test_aws_credentials_get_boto3_session():
-    """
-    Asserts that instantiated AwsCredentials block creates an
-    authenticated boto3 session.
-    """
+if PYDANTIC_VERSION.startswith("2."):
+    from pydantic.v1 import Field, SecretStr
+else:
+    from pydantic import Field, SecretStr
 
-    with mock_s3():
-        aws_credentials_block = AwsCredentials()
-        boto3_session = aws_credentials_block.get_boto3_session()
-        assert isinstance(boto3_session, Session)
+from prefect_aws.client_parameters import AwsClientParameters
 
 
-def test_minio_credentials_get_boto3_session():
-    """
-    Asserts that instantiated MinIOCredentials block creates
-    an authenticated boto3 session.
+class ClientType(Enum):
+    S3 = "s3"
+    ECS = "ecs"
+    BATCH = "batch"
+    SECRETS_MANAGER = "secretsmanager"
+
+
+class AwsCredentials(CredentialsBlock):
     """
+    Block used to manage authentication with AWS. AWS authentication is
+    handled via the `boto3` module. Refer to the
+    [boto3 docs](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html)
+    for more info about the possible credential configurations.
+
+    Example:
+        Load stored AWS credentials:
+        ```python
+        from prefect_aws import AwsCredentials
+
+        aws_credentials_block = AwsCredentials.load("BLOCK_NAME")
+        ```
+    """  # noqa E501
+
+    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png"  # noqa
+    _block_type_name = "AWS Credentials"
+    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.AwsCredentials"  # noqa
+
+    aws_access_key_id: Optional[str] = Field(
+        default=None,
+        description="A specific AWS access key ID.",
+        title="AWS Access Key ID",
+    )
+    aws_secret_access_key: Optional[SecretStr] = Field(
+        default=None,
+        description="A specific AWS secret access key.",
+        title="AWS Access Key Secret",
+    )
+    aws_session_token: Optional[str] = Field(
+        default=None,
+        description=(
+            "The session key for your AWS account. "
+            "This is only needed when you are using temporary credentials."
+        ),
+        title="AWS Session Token",
+    )
+    profile_name: Optional[str] = Field(
+        default=None, description="The profile to use when creating your session."
+    )
+    region_name: Optional[str] = Field(
+        default=None,
+        description="The AWS Region where you want to create new connections.",
+    )
+    aws_client_parameters: AwsClientParameters = Field(
+        default_factory=AwsClientParameters,
+        description="Extra parameters to initialize the Client.",
+        title="AWS Client Parameters",
+    )
 
-    minio_credentials_block = MinIOCredentials(
-        minio_root_user="root_user", minio_root_password="root_password"
+    def get_boto3_session(self) -> boto3.Session:
+        """
+        Returns an authenticated boto3 session that can be used to create clients
+        for AWS services
+
+        Example:
+            Create an S3 client from an authorized boto3 session:
+            ```python
+            aws_credentials = AwsCredentials(
+                aws_access_key_id = "access_key_id",
+                aws_secret_access_key = "secret_access_key"
+                )
+            s3_client = aws_credentials.get_boto3_session().client("s3")
+            ```
+        """
+
+        if self.aws_secret_access_key:
+            aws_secret_access_key = self.aws_secret_access_key.get_secret_value()
+        else:
+            aws_secret_access_key = None
+
+        return boto3.Session(
+            aws_access_key_id=self.aws_access_key_id,
+            aws_secret_access_key=aws_secret_access_key,
+            aws_session_token=self.aws_session_token,
+            profile_name=self.profile_name,
+            region_name=self.region_name,
+        )
+
+    def get_client(self, client_type: Union[str, ClientType]) -> Any:
+        """
+        Helper method to dynamically get a client type.
+
+        Args:
+            client_type: The client's service name.
+
+        Returns:
+            An authenticated client.
+
+        Raises:
+            ValueError: if the client is not supported.
+        """
+        if isinstance(client_type, ClientType):
+            client_type = client_type.value
+
+        client = self.get_boto3_session().client(
+            service_name=client_type, **self.aws_client_parameters.get_params_override()
+        )
+        return client
+
+    def get_s3_client(self) -> S3Client:
+        """
+        Gets an authenticated S3 client.
+
+        Returns:
+            An authenticated S3 client.
+        """
+        return self.get_client(client_type=ClientType.S3)
+
+    def get_secrets_manager_client(self) -> SecretsManagerClient:
+        """
+        Gets an authenticated Secrets Manager client.
+
+        Returns:
+            An authenticated Secrets Manager client.
+        """
+        return self.get_client(client_type=ClientType.SECRETS_MANAGER)
+
+
+class MinIOCredentials(CredentialsBlock):
+    """
+    Block used to manage authentication with MinIO. Refer to the
+    [MinIO docs](https://docs.min.io/docs/minio-server-configuration-guide.html)
+    for more info about the possible credential configurations.
+
+    Attributes:
+        minio_root_user: Admin or root user.
+        minio_root_password: Admin or root password.
+        region_name: Location of server, e.g. "us-east-1".
+
+    Example:
+        Load stored MinIO credentials:
+        ```python
+        from prefect_aws import MinIOCredentials
+
+        minio_credentials_block = MinIOCredentials.load("BLOCK_NAME")
+        ```
+    """  # noqa E501
+
+    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/676cb17bcbdff601f97e0a02ff8bcb480e91ff40-250x250.png"  # noqa
+    _block_type_name = "MinIO Credentials"
+    _description = (
+        "Block used to manage authentication with MinIO. Refer to the MinIO "
+        "docs: https://docs.min.io/docs/minio-server-configuration-guide.html "
+        "for more info about the possible credential configurations."
     )
-    boto3_session = minio_credentials_block.get_boto3_session()
-    assert isinstance(boto3_session, Session)
+    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.MinIOCredentials"  # noqa
 
+    minio_root_user: str = Field(default=..., description="Admin or root user.")
+    minio_root_password: SecretStr = Field(
+        default=..., description="Admin or root password."
+    )
+    region_name: Optional[str] = Field(
+        default=None,
+        description="The AWS Region where you want to create new connections.",
+    )
+    aws_client_parameters: AwsClientParameters = Field(
+        default_factory=AwsClientParameters,
+        description="Extra parameters to initialize the Client.",
+    )
 
-@pytest.mark.parametrize(
-    "credentials",
-    [
-        AwsCredentials(),
-        MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
-        ),
-    ],
-)
-@pytest.mark.parametrize("client_type", ["s3", ClientType.S3])
-def test_credentials_get_client_s3(credentials, client_type):
-    with mock_s3():
-        assert isinstance(credentials.get_client(client_type), BaseClient)
-
-
-@pytest.mark.parametrize(
-    "credentials",
-    [
-        AwsCredentials(),
-        MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
-        ),
-    ],
-)
-@pytest.mark.parametrize("client_type", ["ecs", ClientType.ECS])
-def test_credentials_get_client_ecs(credentials, client_type):
-    with mock_s3():
-        assert isinstance(credentials.get_client(client_type), BaseClient)
-
-
-@pytest.mark.parametrize(
-    "credentials",
-    [
-        AwsCredentials(),
-        MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
-        ),
-    ],
-)
-@pytest.mark.parametrize("client_type", ["batch", ClientType.BATCH])
-def test_credentials_get_client_batch(credentials, client_type):
-    with mock_s3():
-        assert isinstance(credentials.get_client(client_type), BaseClient)
-
-
-@pytest.mark.parametrize(
-    "credentials",
-    [
-        AwsCredentials(),
-        MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
-        ),
-    ],
-)
-@pytest.mark.parametrize("client_type", ["s3", ClientType.S3])
-def test_credentials_get_client_cached_s3(credentials, client_type):
-    with mock_s3():
-        assert isinstance(_get_client_cached(credentials, client_type), BaseClient)
-
-
-@pytest.mark.parametrize(
-    "credentials",
-    [
-        AwsCredentials(),
-        MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
-        ),
-    ],
-)
-@pytest.mark.parametrize("client_type", ["ecs", ClientType.ECS])
-def test_credentials_get_client_cached_ecs(credentials, client_type):
-    with mock_s3():
-        client = _get_client_cached(credentials, client_type)
-
-        assert isinstance(client, BaseClient)
+    def get_boto3_session(self) -> boto3.Session:
+        """
+        Returns an authenticated boto3 session that can be used to create clients
+        and perform object operations on MinIO server.
+
+        Example:
+            Create an S3 client from an authorized boto3 session
+
+            ```python
+            minio_credentials = MinIOCredentials(
+                minio_root_user = "minio_root_user",
+                minio_root_password = "minio_root_password"
+            )
+            s3_client = minio_credentials.get_boto3_session().client(
+                service="s3",
+                endpoint_url="http://localhost:9000"
+            )
+            ```
+        """
+
+        minio_root_password = (
+            self.minio_root_password.get_secret_value()
+            if self.minio_root_password
+            else None
+        )
+
+        return boto3.Session(
+            aws_access_key_id=self.minio_root_user,
+            aws_secret_access_key=minio_root_password,
+            region_name=self.region_name,
+        )
+
+    def get_client(self, client_type: Union[str, ClientType]) -> Any:
+        """
+        Helper method to dynamically get a client type.
+
+        Args:
+            client_type: The client's service name.
+
+        Returns:
+            An authenticated client.
+
+        Raises:
+            ValueError: if the client is not supported.
+        """
+        if isinstance(client_type, ClientType):
+            client_type = client_type.value
+
+        client = self.get_boto3_session().client(
+            service_name=client_type, **self.aws_client_parameters.get_params_override()
+        )
+        return client
+
+    def get_s3_client(self) -> S3Client:
+        """
+        Gets an authenticated S3 client.
+
+        Returns:
+            An authenticated S3 client.
+        """
+        return self.get_client(client_type=ClientType.S3)

From 7428ae33da13e6e4b0b872f04924cacf71233329 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 16:14:17 +0100
Subject: [PATCH 10/30] revert changes

---
 tests/test_credentials.py | 271 +++++---------------------------------
 1 file changed, 34 insertions(+), 237 deletions(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 64f49efe..6e0a1ff8 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -1,249 +1,46 @@
-"""Module handling AWS credentials"""
+import pytest
+from boto3.session import Session
+from botocore.client import BaseClient
+from moto import mock_s3
 
-from enum import Enum
-from typing import Any, Optional, Union
+from prefect_aws.credentials import AwsCredentials, ClientType, MinIOCredentials
 
-import boto3
-from mypy_boto3_s3 import S3Client
-from mypy_boto3_secretsmanager import SecretsManagerClient
-from prefect.blocks.abstract import CredentialsBlock
-from pydantic import VERSION as PYDANTIC_VERSION
 
-if PYDANTIC_VERSION.startswith("2."):
-    from pydantic.v1 import Field, SecretStr
-else:
-    from pydantic import Field, SecretStr
-
-from prefect_aws.client_parameters import AwsClientParameters
-
-
-class ClientType(Enum):
-    S3 = "s3"
-    ECS = "ecs"
-    BATCH = "batch"
-    SECRETS_MANAGER = "secretsmanager"
-
-
-class AwsCredentials(CredentialsBlock):
+def test_aws_credentials_get_boto3_session():
     """
-    Block used to manage authentication with AWS. AWS authentication is
-    handled via the `boto3` module. Refer to the
-    [boto3 docs](https://boto3.amazonaws.com/v1/documentation/api/latest/guide/credentials.html)
-    for more info about the possible credential configurations.
-
-    Example:
-        Load stored AWS credentials:
-        ```python
-        from prefect_aws import AwsCredentials
-
-        aws_credentials_block = AwsCredentials.load("BLOCK_NAME")
-        ```
-    """  # noqa E501
-
-    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/d74b16fe84ce626345adf235a47008fea2869a60-225x225.png"  # noqa
-    _block_type_name = "AWS Credentials"
-    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.AwsCredentials"  # noqa
-
-    aws_access_key_id: Optional[str] = Field(
-        default=None,
-        description="A specific AWS access key ID.",
-        title="AWS Access Key ID",
-    )
-    aws_secret_access_key: Optional[SecretStr] = Field(
-        default=None,
-        description="A specific AWS secret access key.",
-        title="AWS Access Key Secret",
-    )
-    aws_session_token: Optional[str] = Field(
-        default=None,
-        description=(
-            "The session key for your AWS account. "
-            "This is only needed when you are using temporary credentials."
-        ),
-        title="AWS Session Token",
-    )
-    profile_name: Optional[str] = Field(
-        default=None, description="The profile to use when creating your session."
-    )
-    region_name: Optional[str] = Field(
-        default=None,
-        description="The AWS Region where you want to create new connections.",
-    )
-    aws_client_parameters: AwsClientParameters = Field(
-        default_factory=AwsClientParameters,
-        description="Extra parameters to initialize the Client.",
-        title="AWS Client Parameters",
-    )
-
-    def get_boto3_session(self) -> boto3.Session:
-        """
-        Returns an authenticated boto3 session that can be used to create clients
-        for AWS services
-
-        Example:
-            Create an S3 client from an authorized boto3 session:
-            ```python
-            aws_credentials = AwsCredentials(
-                aws_access_key_id = "access_key_id",
-                aws_secret_access_key = "secret_access_key"
-                )
-            s3_client = aws_credentials.get_boto3_session().client("s3")
-            ```
-        """
-
-        if self.aws_secret_access_key:
-            aws_secret_access_key = self.aws_secret_access_key.get_secret_value()
-        else:
-            aws_secret_access_key = None
-
-        return boto3.Session(
-            aws_access_key_id=self.aws_access_key_id,
-            aws_secret_access_key=aws_secret_access_key,
-            aws_session_token=self.aws_session_token,
-            profile_name=self.profile_name,
-            region_name=self.region_name,
-        )
-
-    def get_client(self, client_type: Union[str, ClientType]) -> Any:
-        """
-        Helper method to dynamically get a client type.
-
-        Args:
-            client_type: The client's service name.
-
-        Returns:
-            An authenticated client.
-
-        Raises:
-            ValueError: if the client is not supported.
-        """
-        if isinstance(client_type, ClientType):
-            client_type = client_type.value
-
-        client = self.get_boto3_session().client(
-            service_name=client_type, **self.aws_client_parameters.get_params_override()
-        )
-        return client
-
-    def get_s3_client(self) -> S3Client:
-        """
-        Gets an authenticated S3 client.
-
-        Returns:
-            An authenticated S3 client.
-        """
-        return self.get_client(client_type=ClientType.S3)
-
-    def get_secrets_manager_client(self) -> SecretsManagerClient:
-        """
-        Gets an authenticated Secrets Manager client.
-
-        Returns:
-            An authenticated Secrets Manager client.
-        """
-        return self.get_client(client_type=ClientType.SECRETS_MANAGER)
-
-
-class MinIOCredentials(CredentialsBlock):
+    Asserts that instantiated AwsCredentials block creates an
+    authenticated boto3 session.
     """
-    Block used to manage authentication with MinIO. Refer to the
-    [MinIO docs](https://docs.min.io/docs/minio-server-configuration-guide.html)
-    for more info about the possible credential configurations.
-
-    Attributes:
-        minio_root_user: Admin or root user.
-        minio_root_password: Admin or root password.
-        region_name: Location of server, e.g. "us-east-1".
 
-    Example:
-        Load stored MinIO credentials:
-        ```python
-        from prefect_aws import MinIOCredentials
+    with mock_s3():
+        aws_credentials_block = AwsCredentials()
+        boto3_session = aws_credentials_block.get_boto3_session()
+        assert isinstance(boto3_session, Session)
 
-        minio_credentials_block = MinIOCredentials.load("BLOCK_NAME")
-        ```
-    """  # noqa E501
 
-    _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/676cb17bcbdff601f97e0a02ff8bcb480e91ff40-250x250.png"  # noqa
-    _block_type_name = "MinIO Credentials"
-    _description = (
-        "Block used to manage authentication with MinIO. Refer to the MinIO "
-        "docs: https://docs.min.io/docs/minio-server-configuration-guide.html "
-        "for more info about the possible credential configurations."
-    )
-    _documentation_url = "https://prefecthq.github.io/prefect-aws/credentials/#prefect_aws.credentials.MinIOCredentials"  # noqa
+def test_minio_credentials_get_boto3_session():
+    """
+    Asserts that instantiated MinIOCredentials block creates
+    an authenticated boto3 session.
+    """
 
-    minio_root_user: str = Field(default=..., description="Admin or root user.")
-    minio_root_password: SecretStr = Field(
-        default=..., description="Admin or root password."
-    )
-    region_name: Optional[str] = Field(
-        default=None,
-        description="The AWS Region where you want to create new connections.",
+    minio_credentials_block = MinIOCredentials(
+        minio_root_user="root_user", minio_root_password="root_password"
     )
-    aws_client_parameters: AwsClientParameters = Field(
-        default_factory=AwsClientParameters,
-        description="Extra parameters to initialize the Client.",
-    )
-
-    def get_boto3_session(self) -> boto3.Session:
-        """
-        Returns an authenticated boto3 session that can be used to create clients
-        and perform object operations on MinIO server.
-
-        Example:
-            Create an S3 client from an authorized boto3 session
-
-            ```python
-            minio_credentials = MinIOCredentials(
-                minio_root_user = "minio_root_user",
-                minio_root_password = "minio_root_password"
-            )
-            s3_client = minio_credentials.get_boto3_session().client(
-                service="s3",
-                endpoint_url="http://localhost:9000"
-            )
-            ```
-        """
-
-        minio_root_password = (
-            self.minio_root_password.get_secret_value()
-            if self.minio_root_password
-            else None
-        )
-
-        return boto3.Session(
-            aws_access_key_id=self.minio_root_user,
-            aws_secret_access_key=minio_root_password,
-            region_name=self.region_name,
-        )
+    boto3_session = minio_credentials_block.get_boto3_session()
+    assert isinstance(boto3_session, Session)
 
-    def get_client(self, client_type: Union[str, ClientType]) -> Any:
-        """
-        Helper method to dynamically get a client type.
 
-        Args:
-            client_type: The client's service name.
-
-        Returns:
-            An authenticated client.
-
-        Raises:
-            ValueError: if the client is not supported.
-        """
-        if isinstance(client_type, ClientType):
-            client_type = client_type.value
-
-        client = self.get_boto3_session().client(
-            service_name=client_type, **self.aws_client_parameters.get_params_override()
-        )
-        return client
-
-    def get_s3_client(self) -> S3Client:
-        """
-        Gets an authenticated S3 client.
-
-        Returns:
-            An authenticated S3 client.
-        """
-        return self.get_client(client_type=ClientType.S3)
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
+@pytest.mark.parametrize("client_type", ["s3", ClientType.S3])
+def test_credentials_get_client(credentials, client_type):
+    with mock_s3():
+        assert isinstance(credentials.get_client(client_type), BaseClient)

From a1a8866879fb3c679d2fa1b409fa0610bfc18929 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 16:16:44 +0100
Subject: [PATCH 11/30] fix docs

---
 prefect_aws/credentials.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 0ebfe787..7ad7dc69 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -30,7 +30,7 @@ class ClientType(Enum):
 @lru_cache
 def _get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
     """
-    Helper method to cache and  dynamically get a client type.
+    Helper method to cache and dynamically get a client type.
 
     Args:
         client_type: The client's service name.

From dd0cca41e9322fb51a9e6eb0f9c8a8ce2f4bca9b Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 16:43:15 +0100
Subject: [PATCH 12/30] Update CHANGELOG.md

---
 CHANGELOG.md | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cefe0b6d..0f418adf 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,6 +11,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Handle `boto3` clients more efficiently with `lru_cache` - [#361](https://github.com/PrefectHQ/prefect-aws/pull/361)
+
 ### Fixed
 
 - Bug where `S3Bucket.load()` constructed `AwsCredentials` instead of `MinIOCredentials` - [#359](https://github.com/PrefectHQ/prefect-aws/pull/359)
@@ -95,6 +97,7 @@ Released August 31st, 2023.
 Released July 20th, 2023.
 
 ### Changed
+
 - Promoted workers to GA, removed beta disclaimers
 
 ## 0.3.5
@@ -283,6 +286,7 @@ Released on October 28th, 2022.
 - `ECSTask` is no longer experimental — [#137](https://github.com/PrefectHQ/prefect-aws/pull/137)
 
 ### Fixed
+
 - Fix ignore_file option in `S3Bucket` skipping files which should be included — [#139](https://github.com/PrefectHQ/prefect-aws/pull/139)
 - Fixed bug where `basepath` is used twice in the path when using `S3Bucket.put_directory` - [#143](https://github.com/PrefectHQ/prefect-aws/pull/143)
 

From bcf2bed320e636ae6247a58bdf9ed87ac556dc62 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Fri, 22 Dec 2023 17:04:55 +0100
Subject: [PATCH 13/30] Add maxsize and typed=True

---
 prefect_aws/credentials.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 7ad7dc69..fe497be1 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -27,7 +27,7 @@ class ClientType(Enum):
     SECRETS_MANAGER = "secretsmanager"
 
 
-@lru_cache
+@lru_cache(maxsize=8, typed=True)
 def _get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
     """
     Helper method to cache and dynamically get a client type.
@@ -103,7 +103,7 @@ class AwsCredentials(CredentialsBlock):
     )
 
     class Config:
-        """pydantic config"""
+        """Config class for pydantic model."""
 
         arbitrary_types_allowed = True
 
@@ -216,7 +216,7 @@ class MinIOCredentials(CredentialsBlock):
     )
 
     class Config:
-        """pydantic config"""
+        """Config class for pydantic model."""
 
         arbitrary_types_allowed = True
 

From ab03c45f53adb87c6998a2fe12845d532f83286c Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Wed, 3 Jan 2024 16:46:41 +0100
Subject: [PATCH 14/30] add test

---
 tests/test_credentials.py | 33 ++++++++++++++++++++++++++++++++-
 1 file changed, 32 insertions(+), 1 deletion(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 6e0a1ff8..85168741 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -1,9 +1,16 @@
+from unittest.mock import patch
+
 import pytest
 from boto3.session import Session
 from botocore.client import BaseClient
 from moto import mock_s3
 
-from prefect_aws.credentials import AwsCredentials, ClientType, MinIOCredentials
+from prefect_aws.credentials import (
+    AwsCredentials,
+    ClientType,
+    MinIOCredentials,
+    _get_client_cached,
+)
 
 
 def test_aws_credentials_get_boto3_session():
@@ -44,3 +51,27 @@ def test_minio_credentials_get_boto3_session():
 def test_credentials_get_client(credentials, client_type):
     with mock_s3():
         assert isinstance(credentials.get_client(client_type), BaseClient)
+
+
+@patch("prefect_aws.credentials._get_client_cached")
+def test_get_client_cached(mock_get_client_cached):
+    """
+    Test to ensure that _get_client_cached function returns the same instance
+    for multiple calls with the same parameters and properly utilizes lru_cache.
+    """
+
+    # Create a mock AwsCredentials instance
+    aws_credentials_block = AwsCredentials()
+
+    # Call _get_client_cached multiple times with the same parameters
+    _get_client_cached(aws_credentials_block, ClientType.S3)
+    _get_client_cached(aws_credentials_block, ClientType.S3)
+
+    # Verify that _get_client_cached is called only once due to caching
+    mock_get_client_cached.assert_called_once_with(aws_credentials_block, ClientType.S3)
+
+    # Test with different parameters to ensure they are cached separately
+    _get_client_cached(aws_credentials_block, ClientType.ECS)
+    assert (
+        mock_get_client_cached.call_count == 2
+    ), "Should be called twice with different parameters"

From 59e38d1fef89706df87f323d3ea7f747aa7eee3f Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Wed, 3 Jan 2024 18:00:06 +0100
Subject: [PATCH 15/30] Test with cache_info

---
 tests/test_credentials.py | 32 ++++++++++++++++++--------------
 1 file changed, 18 insertions(+), 14 deletions(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 85168741..52e6f91e 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -1,5 +1,3 @@
-from unittest.mock import patch
-
 import pytest
 from boto3.session import Session
 from botocore.client import BaseClient
@@ -53,25 +51,31 @@ def test_credentials_get_client(credentials, client_type):
         assert isinstance(credentials.get_client(client_type), BaseClient)
 
 
-@patch("prefect_aws.credentials._get_client_cached")
-def test_get_client_cached(mock_get_client_cached):
+@pytest.mark.parametrize("credentials", [AwsCredentials()])
+def test_get_client_cached(credentials):
     """
     Test to ensure that _get_client_cached function returns the same instance
     for multiple calls with the same parameters and properly utilizes lru_cache.
     """
 
-    # Create a mock AwsCredentials instance
-    aws_credentials_block = AwsCredentials()
+    # Clear cache
+    _get_client_cached.cache_clear()
+
+    assert _get_client_cached.cache_info().hits == 0, "Initial call count should be 0"
 
-    # Call _get_client_cached multiple times with the same parameters
-    _get_client_cached(aws_credentials_block, ClientType.S3)
-    _get_client_cached(aws_credentials_block, ClientType.S3)
+    # Call get_client multiple times with the same parameters
+    credentials.get_client(ClientType.S3)
+    credentials.get_client(ClientType.S3)
+    credentials.get_client(ClientType.S3)
 
     # Verify that _get_client_cached is called only once due to caching
-    mock_get_client_cached.assert_called_once_with(aws_credentials_block, ClientType.S3)
+    assert _get_client_cached.cache_info().misses == 1
+    assert _get_client_cached.cache_info().hits == 2
 
     # Test with different parameters to ensure they are cached separately
-    _get_client_cached(aws_credentials_block, ClientType.ECS)
-    assert (
-        mock_get_client_cached.call_count == 2
-    ), "Should be called twice with different parameters"
+    credentials.get_client(ClientType.ECS)
+    credentials.get_client(ClientType.ECS)
+
+    # "Should be called again with different parameters"
+    assert _get_client_cached.cache_info().misses == 2
+    assert _get_client_cached.cache_info().hits == 3

From 3cd1eab81958cba92dab94cdbb083a8778c4bd80 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Wed, 3 Jan 2024 18:13:11 +0100
Subject: [PATCH 16/30] Update AwsCredentials

---
 tests/test_credentials.py | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 52e6f91e..85992027 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -51,30 +51,32 @@ def test_credentials_get_client(credentials, client_type):
         assert isinstance(credentials.get_client(client_type), BaseClient)
 
 
-@pytest.mark.parametrize("credentials", [AwsCredentials()])
-def test_get_client_cached(credentials):
+def test_get_client_cached():
     """
     Test to ensure that _get_client_cached function returns the same instance
     for multiple calls with the same parameters and properly utilizes lru_cache.
     """
 
+    # Create a mock AwsCredentials instance
+    aws_credentials_block = AwsCredentials()
+
     # Clear cache
     _get_client_cached.cache_clear()
 
     assert _get_client_cached.cache_info().hits == 0, "Initial call count should be 0"
 
     # Call get_client multiple times with the same parameters
-    credentials.get_client(ClientType.S3)
-    credentials.get_client(ClientType.S3)
-    credentials.get_client(ClientType.S3)
+    aws_credentials_block.get_client(ClientType.S3)
+    aws_credentials_block.get_client(ClientType.S3)
+    aws_credentials_block.get_client(ClientType.S3)
 
     # Verify that _get_client_cached is called only once due to caching
     assert _get_client_cached.cache_info().misses == 1
     assert _get_client_cached.cache_info().hits == 2
 
     # Test with different parameters to ensure they are cached separately
-    credentials.get_client(ClientType.ECS)
-    credentials.get_client(ClientType.ECS)
+    aws_credentials_block.get_client(ClientType.SECRETS_MANAGER)
+    aws_credentials_block.get_client(ClientType.SECRETS_MANAGER)
 
     # "Should be called again with different parameters"
     assert _get_client_cached.cache_info().misses == 2

From e0a927b83d76e71da06d9034c03c56d05544daa6 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Wed, 3 Jan 2024 18:23:23 +0100
Subject: [PATCH 17/30] Only S3

---
 tests/test_credentials.py | 8 --------
 1 file changed, 8 deletions(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 85992027..1e475a85 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -73,11 +73,3 @@ def test_get_client_cached():
     # Verify that _get_client_cached is called only once due to caching
     assert _get_client_cached.cache_info().misses == 1
     assert _get_client_cached.cache_info().hits == 2
-
-    # Test with different parameters to ensure they are cached separately
-    aws_credentials_block.get_client(ClientType.SECRETS_MANAGER)
-    aws_credentials_block.get_client(ClientType.SECRETS_MANAGER)
-
-    # "Should be called again with different parameters"
-    assert _get_client_cached.cache_info().misses == 2
-    assert _get_client_cached.cache_info().hits == 3

From 0de61f81d046c9bfebbcacbc26508a3fd5396c90 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Wed, 3 Jan 2024 18:36:38 +0100
Subject: [PATCH 18/30] Empty-Commit


From 283050ec730e4002cb2208a7174042b401f9c159 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Thu, 18 Jan 2024 18:30:40 +0000
Subject: [PATCH 19/30] Update hash function

---
 prefect_aws/credentials.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index fe497be1..5125e56a 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -108,7 +108,7 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return id(self)
+        return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -221,7 +221,7 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return id(self)
+        return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """

From 3200967f4959f307bbb43b8fbcfb23faae543be2 Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Thu, 18 Jan 2024 19:47:22 +0000
Subject: [PATCH 20/30] Revert changes

---
 prefect_aws/credentials.py | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 5125e56a..8565126c 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -108,7 +108,8 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return hash(self.json())
+        return id(self)
+        # return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -221,7 +222,8 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return hash(self.json())
+        return id(self)
+        # return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """

From e8b61e082377fa380eb76992aeada3f23a8adf6b Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Thu, 18 Jan 2024 19:55:56 +0000
Subject: [PATCH 21/30] Update hash

---
 prefect_aws/credentials.py | 6 ++----
 1 file changed, 2 insertions(+), 4 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 8565126c..5125e56a 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -108,8 +108,7 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return id(self)
-        # return hash(self.json())
+        return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -222,8 +221,7 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return id(self)
-        # return hash(self.json())
+        return hash(self.json())
 
     def get_boto3_session(self) -> boto3.Session:
         """

From 56210f8caa4c4e62f2aeff0bbf7ac2886261c21e Mon Sep 17 00:00:00 2001
From: Mattia <5013654+mattiamatrix@users.noreply.github.com>
Date: Thu, 18 Jan 2024 20:58:22 +0000
Subject: [PATCH 22/30] Test different hash

---
 prefect_aws/client_parameters.py | 12 ++++++++++++
 prefect_aws/credentials.py       | 20 ++++++++++++++++++--
 2 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/prefect_aws/client_parameters.py b/prefect_aws/client_parameters.py
index bf030590..eb3be09b 100644
--- a/prefect_aws/client_parameters.py
+++ b/prefect_aws/client_parameters.py
@@ -70,6 +70,18 @@ class AwsClientParameters(BaseModel):
         title="Botocore Config",
     )
 
+    def __hash__(self):
+        return hash(
+            (
+                self.api_version,
+                self.use_ssl,
+                self.verify,
+                self.verify_cert_path,
+                self.endpoint_url,
+                self.config,
+            )
+        )
+
     @validator("config", pre=True)
     def instantiate_config(cls, value: Union[Config, Dict[str, Any]]) -> Dict[str, Any]:
         """
diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 5125e56a..0646f1b4 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -108,7 +108,16 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return hash(self.json())
+        return hash(
+            (
+                self.aws_access_key_id,
+                self.aws_secret_access_key,
+                self.aws_session_token,
+                self.profile_name,
+                self.region_name,
+                self.aws_client_parameters,
+            )
+        )
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -221,7 +230,14 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return hash(self.json())
+        return hash(
+            (
+                self.minio_root_user,
+                self.minio_root_password,
+                self.region_name,
+                self.aws_client_parameters,
+            )
+        )
 
     def get_boto3_session(self) -> boto3.Session:
         """

From 865975ca1563475e529fd6505f908beb9fe7d5d9 Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Thu, 18 Jan 2024 17:03:00 -0600
Subject: [PATCH 23/30] avoid modifying default behavior

---
 prefect_aws/credentials.py | 22 ++++++++++++++++++++--
 prefect_aws/s3.py          | 11 ++++++++++-
 tests/test_credentials.py  | 18 ++++++++++++------
 3 files changed, 42 insertions(+), 9 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 0646f1b4..14854e21 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -148,7 +148,7 @@ def get_boto3_session(self) -> boto3.Session:
             region_name=self.region_name,
         )
 
-    def get_client(self, client_type: Union[str, ClientType]) -> Any:
+    def get_client(self, client_type: Union[str, ClientType], use_cache: bool = False):
         """
         Helper method to dynamically get a client type.
 
@@ -161,6 +161,15 @@ def get_client(self, client_type: Union[str, ClientType]) -> Any:
         Raises:
             ValueError: if the client is not supported.
         """
+        if isinstance(client_type, ClientType):
+            client_type = client_type.value
+
+        if not use_cache:
+            return self.get_boto3_session().client(
+                service_name=client_type,
+                **self.aws_client_parameters.get_params_override()
+            )
+    
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
@@ -271,7 +280,7 @@ def get_boto3_session(self) -> boto3.Session:
             region_name=self.region_name,
         )
 
-    def get_client(self, client_type: Union[str, ClientType]) -> Any:
+    def get_client(self, client_type: Union[str, ClientType], use_cache: bool = False):
         """
         Helper method to dynamically get a client type.
 
@@ -284,6 +293,15 @@ def get_client(self, client_type: Union[str, ClientType]) -> Any:
         Raises:
             ValueError: if the client is not supported.
         """
+        if isinstance(client_type, ClientType):
+            client_type = client_type.value
+
+        if not use_cache:
+            return self.get_boto3_session().client(
+                service_name=client_type,
+                **self.aws_client_parameters.get_params_override()
+            )
+    
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 643d78ac..32e6c044 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -424,6 +424,15 @@ class S3Bucket(WritableFileSystem, WritableDeploymentStorage, ObjectStorageBlock
             "for reading and writing objects."
         ),
     )
+    
+    cache_client: bool = Field(
+        default=False,
+        description=(
+            "If True, the S3 client will be cached. This is useful for "
+            "performance, but can cause issues if the S3 client is used "
+            "in multiple threads."
+        ),
+    )
 
     # Property to maintain compatibility with storage block based deployments
     @property
@@ -466,7 +475,7 @@ def _get_s3_client(self) -> boto3.client:
         Authenticate MinIO credentials or AWS credentials and return an S3 client.
         This is a helper function called by read_path() or write_path().
         """
-        return self.credentials.get_s3_client()
+        return self.credentials.get_client("s3", use_cache=self.cache_client)
 
     def _get_bucket_resource(self) -> boto3.resource:
         """
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 1e475a85..2a2ce072 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -50,8 +50,10 @@ def test_credentials_get_client(credentials, client_type):
     with mock_s3():
         assert isinstance(credentials.get_client(client_type), BaseClient)
 
-
-def test_get_client_cached():
+@pytest.mark.parametrize(
+    "client_type", [member.value for member in ClientType]
+)
+def test_get_client_cached(client_type):
     """
     Test to ensure that _get_client_cached function returns the same instance
     for multiple calls with the same parameters and properly utilizes lru_cache.
@@ -65,11 +67,15 @@ def test_get_client_cached():
 
     assert _get_client_cached.cache_info().hits == 0, "Initial call count should be 0"
 
+    assert aws_credentials_block.get_client(client_type) is not None
+    
+    assert _get_client_cached.cache_info().hits == 0, "Cache should not yet be used"
+
     # Call get_client multiple times with the same parameters
-    aws_credentials_block.get_client(ClientType.S3)
-    aws_credentials_block.get_client(ClientType.S3)
-    aws_credentials_block.get_client(ClientType.S3)
+    aws_credentials_block.get_client(client_type, use_cache=True)
+    aws_credentials_block.get_client(client_type, use_cache=True)
+    aws_credentials_block.get_client(client_type, use_cache=True)
 
     # Verify that _get_client_cached is called only once due to caching
     assert _get_client_cached.cache_info().misses == 1
-    assert _get_client_cached.cache_info().hits == 2
+    assert _get_client_cached.cache_info().hits == 2
\ No newline at end of file

From 9b6cd9b6a221f5af4402ae73b805cc71fc9a20fc Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Thu, 18 Jan 2024 17:10:31 -0600
Subject: [PATCH 24/30] run pre-commits

---
 prefect_aws/credentials.py | 8 ++++----
 prefect_aws/s3.py          | 2 +-
 tests/test_credentials.py  | 9 ++++-----
 3 files changed, 9 insertions(+), 10 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 14854e21..7845b74b 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -167,9 +167,9 @@ def get_client(self, client_type: Union[str, ClientType], use_cache: bool = Fals
         if not use_cache:
             return self.get_boto3_session().client(
                 service_name=client_type,
-                **self.aws_client_parameters.get_params_override()
+                **self.aws_client_parameters.get_params_override(),
             )
-    
+
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
@@ -299,9 +299,9 @@ def get_client(self, client_type: Union[str, ClientType], use_cache: bool = Fals
         if not use_cache:
             return self.get_boto3_session().client(
                 service_name=client_type,
-                **self.aws_client_parameters.get_params_override()
+                **self.aws_client_parameters.get_params_override(),
             )
-    
+
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 32e6c044..c9d63113 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -424,7 +424,7 @@ class S3Bucket(WritableFileSystem, WritableDeploymentStorage, ObjectStorageBlock
             "for reading and writing objects."
         ),
     )
-    
+
     cache_client: bool = Field(
         default=False,
         description=(
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 2a2ce072..2159c6f4 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -50,9 +50,8 @@ def test_credentials_get_client(credentials, client_type):
     with mock_s3():
         assert isinstance(credentials.get_client(client_type), BaseClient)
 
-@pytest.mark.parametrize(
-    "client_type", [member.value for member in ClientType]
-)
+
+@pytest.mark.parametrize("client_type", [member.value for member in ClientType])
 def test_get_client_cached(client_type):
     """
     Test to ensure that _get_client_cached function returns the same instance
@@ -68,7 +67,7 @@ def test_get_client_cached(client_type):
     assert _get_client_cached.cache_info().hits == 0, "Initial call count should be 0"
 
     assert aws_credentials_block.get_client(client_type) is not None
-    
+
     assert _get_client_cached.cache_info().hits == 0, "Cache should not yet be used"
 
     # Call get_client multiple times with the same parameters
@@ -78,4 +77,4 @@ def test_get_client_cached(client_type):
 
     # Verify that _get_client_cached is called only once due to caching
     assert _get_client_cached.cache_info().misses == 1
-    assert _get_client_cached.cache_info().hits == 2
\ No newline at end of file
+    assert _get_client_cached.cache_info().hits == 2

From 1f9c3d0f45b1332e686c8f89b2cb9ea6d462029f Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Thu, 18 Jan 2024 17:20:21 -0600
Subject: [PATCH 25/30] no way

---
 tests/test_credentials.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 2159c6f4..1b66421b 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -59,7 +59,7 @@ def test_get_client_cached(client_type):
     """
 
     # Create a mock AwsCredentials instance
-    aws_credentials_block = AwsCredentials()
+    aws_credentials_block = AwsCredentials(region_name="us-east-1")
 
     # Clear cache
     _get_client_cached.cache_clear()

From 7202545b437e31a88c164f570f71c07d37c0ffe0 Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Thu, 18 Jan 2024 17:46:41 -0600
Subject: [PATCH 26/30] test caching via s3

---
 tests/test_s3.py | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/tests/test_s3.py b/tests/test_s3.py
index 93d11cc1..d6741625 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -12,6 +12,7 @@
 
 from prefect_aws import AwsCredentials, MinIOCredentials
 from prefect_aws.client_parameters import AwsClientParameters
+from prefect_aws.credentials import _get_client_cached
 from prefect_aws.s3 import (
     S3Bucket,
     s3_copy,
@@ -1047,3 +1048,16 @@ def test_move_object_between_buckets(
 
         with pytest.raises(ClientError):
             assert s3_bucket_with_object.read_path("object") == b"TEST"
+
+    def test_client_is_cached_when_specified(self, aws_creds_block):
+        s3_bucket = S3Bucket(
+            bucket_name="bucket", credentials=aws_creds_block, cache_client=True
+        )
+
+        _get_client_cached.cache_clear()
+
+        s3_bucket._get_s3_client()
+        s3_bucket._get_s3_client()
+
+        assert _get_client_cached.cache_info().hits == 1
+        assert _get_client_cached.cache_info().misses == 1

From 5aad7ab510066c8bd3c475d974af7ed50432413c Mon Sep 17 00:00:00 2001
From: nate nowack <nate@prefect.io>
Date: Thu, 18 Jan 2024 17:58:39 -0600
Subject: [PATCH 27/30] Update prefect_aws/s3.py

---
 prefect_aws/s3.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index c9d63113..34ca076f 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -429,8 +429,8 @@ class S3Bucket(WritableFileSystem, WritableDeploymentStorage, ObjectStorageBlock
         default=False,
         description=(
             "If True, the S3 client will be cached. This is useful for "
-            "performance, but can cause issues if the S3 client is used "
-            "in multiple threads."
+            "performance, but could cause issues if the S3 client is used "
+            "in multi-threaded environments."
         ),
     )
 

From 8379d3dc31c4a906c36cc3d1b3c4c7a526a89d5b Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Fri, 19 Jan 2024 11:36:49 -0600
Subject: [PATCH 28/30] caching by default, remove toggle

---
 prefect_aws/credentials.py | 49 +++++++++------------
 prefect_aws/s3.py          | 11 +----
 tests/test_credentials.py  | 87 ++++++++++++++++++++++++++++++++------
 tests/test_s3.py           | 14 ------
 4 files changed, 94 insertions(+), 67 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 7845b74b..57805d41 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -2,6 +2,7 @@
 
 from enum import Enum
 from functools import lru_cache
+from threading import Lock
 from typing import Any, Optional, Union
 
 import boto3
@@ -17,6 +18,8 @@
 
 from prefect_aws.client_parameters import AwsClientParameters
 
+_LOCK = Lock()
+
 
 class ClientType(Enum):
     """The supported boto3 clients."""
@@ -41,13 +44,14 @@ def _get_client_cached(ctx, client_type: Union[str, ClientType]) -> Any:
     Raises:
         ValueError: if the client is not supported.
     """
-    if isinstance(client_type, ClientType):
-        client_type = client_type.value
+    with _LOCK:
+        if isinstance(client_type, ClientType):
+            client_type = client_type.value
 
-    client = ctx.get_boto3_session().client(
-        service_name=client_type,
-        **ctx.aws_client_parameters.get_params_override(),
-    )
+        client = ctx.get_boto3_session().client(
+            service_name=client_type,
+            **ctx.aws_client_parameters.get_params_override(),
+        )
     return client
 
 
@@ -108,16 +112,15 @@ class Config:
         arbitrary_types_allowed = True
 
     def __hash__(self):
-        return hash(
-            (
-                self.aws_access_key_id,
-                self.aws_secret_access_key,
-                self.aws_session_token,
-                self.profile_name,
-                self.region_name,
-                self.aws_client_parameters,
-            )
+        field_hashes = (
+            hash(self.aws_access_key_id),
+            hash(self.aws_secret_access_key),
+            hash(self.aws_session_token),
+            hash(self.profile_name),
+            hash(self.region_name),
+            hash(frozenset(self.aws_client_parameters.dict().items())),
         )
+        return hash(field_hashes)
 
     def get_boto3_session(self) -> boto3.Session:
         """
@@ -148,7 +151,7 @@ def get_boto3_session(self) -> boto3.Session:
             region_name=self.region_name,
         )
 
-    def get_client(self, client_type: Union[str, ClientType], use_cache: bool = False):
+    def get_client(self, client_type: Union[str, ClientType]):
         """
         Helper method to dynamically get a client type.
 
@@ -164,12 +167,6 @@ def get_client(self, client_type: Union[str, ClientType], use_cache: bool = Fals
         if isinstance(client_type, ClientType):
             client_type = client_type.value
 
-        if not use_cache:
-            return self.get_boto3_session().client(
-                service_name=client_type,
-                **self.aws_client_parameters.get_params_override(),
-            )
-
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
@@ -280,7 +277,7 @@ def get_boto3_session(self) -> boto3.Session:
             region_name=self.region_name,
         )
 
-    def get_client(self, client_type: Union[str, ClientType], use_cache: bool = False):
+    def get_client(self, client_type: Union[str, ClientType]):
         """
         Helper method to dynamically get a client type.
 
@@ -296,12 +293,6 @@ def get_client(self, client_type: Union[str, ClientType], use_cache: bool = Fals
         if isinstance(client_type, ClientType):
             client_type = client_type.value
 
-        if not use_cache:
-            return self.get_boto3_session().client(
-                service_name=client_type,
-                **self.aws_client_parameters.get_params_override(),
-            )
-
         return _get_client_cached(ctx=self, client_type=client_type)
 
     def get_s3_client(self) -> S3Client:
diff --git a/prefect_aws/s3.py b/prefect_aws/s3.py
index 34ca076f..a10e2171 100644
--- a/prefect_aws/s3.py
+++ b/prefect_aws/s3.py
@@ -425,15 +425,6 @@ class S3Bucket(WritableFileSystem, WritableDeploymentStorage, ObjectStorageBlock
         ),
     )
 
-    cache_client: bool = Field(
-        default=False,
-        description=(
-            "If True, the S3 client will be cached. This is useful for "
-            "performance, but could cause issues if the S3 client is used "
-            "in multi-threaded environments."
-        ),
-    )
-
     # Property to maintain compatibility with storage block based deployments
     @property
     def basepath(self) -> str:
@@ -475,7 +466,7 @@ def _get_s3_client(self) -> boto3.client:
         Authenticate MinIO credentials or AWS credentials and return an S3 client.
         This is a helper function called by read_path() or write_path().
         """
-        return self.credentials.get_client("s3", use_cache=self.cache_client)
+        return self.credentials.get_client("s3")
 
     def _get_bucket_resource(self) -> boto3.resource:
         """
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 1b66421b..2f338425 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -51,30 +51,89 @@ def test_credentials_get_client(credentials, client_type):
         assert isinstance(credentials.get_client(client_type), BaseClient)
 
 
+@pytest.mark.parametrize(
+    "credentials",
+    [
+        AwsCredentials(region_name="us-east-1"),
+        MinIOCredentials(
+            minio_root_user="root_user", minio_root_password="root_password"
+        ),
+    ],
+)
 @pytest.mark.parametrize("client_type", [member.value for member in ClientType])
-def test_get_client_cached(client_type):
+def test_get_client_cached(credentials, client_type):
     """
     Test to ensure that _get_client_cached function returns the same instance
     for multiple calls with the same parameters and properly utilizes lru_cache.
     """
 
-    # Create a mock AwsCredentials instance
-    aws_credentials_block = AwsCredentials(region_name="us-east-1")
-
-    # Clear cache
     _get_client_cached.cache_clear()
 
     assert _get_client_cached.cache_info().hits == 0, "Initial call count should be 0"
 
-    assert aws_credentials_block.get_client(client_type) is not None
-
-    assert _get_client_cached.cache_info().hits == 0, "Cache should not yet be used"
-
-    # Call get_client multiple times with the same parameters
-    aws_credentials_block.get_client(client_type, use_cache=True)
-    aws_credentials_block.get_client(client_type, use_cache=True)
-    aws_credentials_block.get_client(client_type, use_cache=True)
+    credentials.get_client(client_type)
+    credentials.get_client(client_type)
+    credentials.get_client(client_type)
 
-    # Verify that _get_client_cached is called only once due to caching
     assert _get_client_cached.cache_info().misses == 1
     assert _get_client_cached.cache_info().hits == 2
+
+
+@pytest.mark.parametrize("client_type", [member.value for member in ClientType])
+def test_aws_credentials_change_causes_cache_miss(client_type):
+    """
+    Test to ensure that changing configuration on an AwsCredentials instance
+    after fetching a client causes a cache miss.
+    """
+
+    _get_client_cached.cache_clear()
+
+    credentials = AwsCredentials(region_name="us-east-1")
+
+    initial_client = credentials.get_client(client_type)
+
+    credentials.region_name = "us-west-2"
+
+    new_client = credentials.get_client(client_type)
+
+    assert (
+        initial_client is not new_client
+    ), "Client should be different after configuration change"
+
+    assert _get_client_cached.cache_info().misses == 2, "Cache should miss twice"
+
+
+@pytest.mark.parametrize("client_type", [member.value for member in ClientType])
+def test_minio_credentials_change_causes_cache_miss(client_type):
+    """
+    Test to ensure that changing configuration on an AwsCredentials instance
+    after fetching a client causes a cache miss.
+    """
+
+    _get_client_cached.cache_clear()
+
+    credentials = MinIOCredentials(
+        minio_root_user="root_user", minio_root_password="root_password"
+    )
+
+    initial_client = credentials.get_client(client_type)
+
+    credentials.region_name = "us-west-2"
+
+    new_client = credentials.get_client(client_type)
+
+    assert (
+        initial_client is not new_client
+    ), "Client should be different after configuration change"
+
+    assert _get_client_cached.cache_info().misses == 2, "Cache should miss twice"
+
+
+def test_aws_credentials_hash_changes():
+    credentials = AwsCredentials(region_name="us-east-1")
+    initial_hash = hash(credentials)
+
+    credentials.region_name = "us-west-2"
+    new_hash = hash(credentials)
+
+    assert initial_hash != new_hash, "Hash should change when region_name changes"
diff --git a/tests/test_s3.py b/tests/test_s3.py
index d6741625..93d11cc1 100644
--- a/tests/test_s3.py
+++ b/tests/test_s3.py
@@ -12,7 +12,6 @@
 
 from prefect_aws import AwsCredentials, MinIOCredentials
 from prefect_aws.client_parameters import AwsClientParameters
-from prefect_aws.credentials import _get_client_cached
 from prefect_aws.s3 import (
     S3Bucket,
     s3_copy,
@@ -1048,16 +1047,3 @@ def test_move_object_between_buckets(
 
         with pytest.raises(ClientError):
             assert s3_bucket_with_object.read_path("object") == b"TEST"
-
-    def test_client_is_cached_when_specified(self, aws_creds_block):
-        s3_bucket = S3Bucket(
-            bucket_name="bucket", credentials=aws_creds_block, cache_client=True
-        )
-
-        _get_client_cached.cache_clear()
-
-        s3_bucket._get_s3_client()
-        s3_bucket._get_s3_client()
-
-        assert _get_client_cached.cache_info().hits == 1
-        assert _get_client_cached.cache_info().misses == 1

From 95df7a0671aaaaed3bf63f29064d798451d31964 Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Fri, 19 Jan 2024 11:50:35 -0600
Subject: [PATCH 29/30] region

---
 prefect_aws/credentials.py | 8 ++++----
 tests/test_credentials.py  | 8 ++++++--
 2 files changed, 10 insertions(+), 6 deletions(-)

diff --git a/prefect_aws/credentials.py b/prefect_aws/credentials.py
index 57805d41..5aeddaa6 100644
--- a/prefect_aws/credentials.py
+++ b/prefect_aws/credentials.py
@@ -238,10 +238,10 @@ class Config:
     def __hash__(self):
         return hash(
             (
-                self.minio_root_user,
-                self.minio_root_password,
-                self.region_name,
-                self.aws_client_parameters,
+                hash(self.minio_root_user),
+                hash(self.minio_root_password),
+                hash(self.region_name),
+                hash(frozenset(self.aws_client_parameters.dict().items())),
             )
         )
 
diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 2f338425..3e98edb0 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -56,7 +56,9 @@ def test_credentials_get_client(credentials, client_type):
     [
         AwsCredentials(region_name="us-east-1"),
         MinIOCredentials(
-            minio_root_user="root_user", minio_root_password="root_password"
+            minio_root_user="root_user",
+            minio_root_password="root_password",
+            region_name="us-east-1",
         ),
     ],
 )
@@ -113,7 +115,9 @@ def test_minio_credentials_change_causes_cache_miss(client_type):
     _get_client_cached.cache_clear()
 
     credentials = MinIOCredentials(
-        minio_root_user="root_user", minio_root_password="root_password"
+        minio_root_user="root_user",
+        minio_root_password="root_password",
+        region_name="us-east-1",
     )
 
     initial_client = credentials.get_client(client_type)

From 1fde31e8bb59ab3c299e7c3e7dbdfe9a540f5e1f Mon Sep 17 00:00:00 2001
From: Nathan Nowack <thrast36@gmail.com>
Date: Fri, 19 Jan 2024 11:56:28 -0600
Subject: [PATCH 30/30] check hashing on both creds classes

---
 tests/test_credentials.py | 29 ++++++++++++++++++++++++++---
 1 file changed, 26 insertions(+), 3 deletions(-)

diff --git a/tests/test_credentials.py b/tests/test_credentials.py
index 3e98edb0..96ecbd22 100644
--- a/tests/test_credentials.py
+++ b/tests/test_credentials.py
@@ -133,11 +133,34 @@ def test_minio_credentials_change_causes_cache_miss(client_type):
     assert _get_client_cached.cache_info().misses == 2, "Cache should miss twice"
 
 
-def test_aws_credentials_hash_changes():
-    credentials = AwsCredentials(region_name="us-east-1")
+@pytest.mark.parametrize(
+    "credentials_type, initial_field, new_field",
+    [
+        (
+            AwsCredentials,
+            {"region_name": "us-east-1"},
+            {"region_name": "us-east-2"},
+        ),
+        (
+            MinIOCredentials,
+            {
+                "region_name": "us-east-1",
+                "minio_root_user": "root_user",
+                "minio_root_password": "root_password",
+            },
+            {
+                "region_name": "us-east-2",
+                "minio_root_user": "root_user",
+                "minio_root_password": "root_password",
+            },
+        ),
+    ],
+)
+def test_aws_credentials_hash_changes(credentials_type, initial_field, new_field):
+    credentials = credentials_type(**initial_field)
     initial_hash = hash(credentials)
 
-    credentials.region_name = "us-west-2"
+    setattr(credentials, list(new_field.keys())[0], list(new_field.values())[0])
     new_hash = hash(credentials)
 
     assert initial_hash != new_hash, "Hash should change when region_name changes"