Don't call get_connection from provide_bucket_name

airflow/providers/amazon/aws/hooks/s3.py

@@ -57,10 +57,8 @@ def wrapper(*args, **kwargs) -> T:
 
         if "bucket_name" not in bound_args.arguments:
             self = args[0]
-            if self.aws_conn_id:
-                connection = self.get_connection(self.aws_conn_id)
-                if connection.schema:
-                    bound_args.arguments["bucket_name"] = connection.schema
+            if self.conn_config and self.conn_config.schema:
+                bound_args.arguments["bucket_name"] = self.conn_config.schema
 
         return func(*bound_args.args, **bound_args.kwargs)
 

airflow/providers/amazon/aws/utils/connection_wrapper.py

@@ -103,6 +103,7 @@ class AwsConnectionWrapper(LoggingMixin):
     conn_type: str | None = field(init=False, default=None)
     login: str | None = field(init=False, repr=False, default=None)
     password: str | None = field(init=False, repr=False, default=None)
+    schema: str | None = field(init=False, repr=False, default=None)
     extra_config: dict[str, Any] = field(init=False, repr=False, default_factory=dict)
 
     # AWS Credentials from connection.
@@ -156,6 +157,7 @@ def __post_init__(self, conn: Connection):
         self.conn_type = conn.conn_type or "aws"
         self.login = conn.login
         self.password = conn.password
+        self.schema = conn.schema or None
         self.extra_config = deepcopy(conn.extra_dejson)
 
         if self.conn_type.lower() == "s3":

tests/providers/amazon/aws/hooks/test_s3.py

@@ -472,13 +472,9 @@ class FakeS3Hook(S3Hook):
             def test_function(self, bucket_name=None):
                 return bucket_name
 
-        fake_s3_hook = FakeS3Hook()
-
-        test_bucket_name = fake_s3_hook.test_function()
-        assert test_bucket_name == mock_get_connection.return_value.schema
-
-        test_bucket_name = fake_s3_hook.test_function(bucket_name="bucket")
-        assert test_bucket_name == "bucket"
+        hook = FakeS3Hook()
+        assert hook.test_function() == "test_bucket"
+        assert hook.test_function(bucket_name="bucket") == "bucket"
 
     def test_delete_objects_key_does_not_exist(self, s3_bucket):
         # The behaviour of delete changed in recent version of s3 mock libraries.

tests/providers/amazon/aws/utils/test_connection_wrapper.py

@@ -89,10 +89,10 @@ def test_values_from_connection(self, extra):
         assert wrap_conn.extra_config is not mock_conn.extra_dejson
         # `extra_config` is a same object that return by `extra_dejson`
         assert wrap_conn.extra_config is wrap_conn.extra_dejson
+        assert wrap_conn.schema == "mock-schema"
 
         # Check that not assigned other attributes from airflow.models.Connection to wrapper
         assert not hasattr(wrap_conn, "host")
-        assert not hasattr(wrap_conn, "schema")
         assert not hasattr(wrap_conn, "port")
 
         # Check that Wrapper is True if assign connection