allow custom session, re-use session if it exists

Author: jayceslesar

pyiceberg/catalog/rest/__init__.py

@@ -32,6 +32,7 @@
 
 from pyiceberg import __version__
 from pyiceberg.catalog import (
+    BOTOCORE_SESSION,
     TOKEN,
     URI,
     WAREHOUSE_LOCATION,
@@ -53,6 +54,7 @@
     TableAlreadyExistsError,
     UnauthorizedError,
 )
+from pyiceberg.io import AWS_ACCESS_KEY_ID, AWS_REGION, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN
 from pyiceberg.partitioning import UNPARTITIONED_PARTITION_SPEC, PartitionSpec, assign_fresh_partition_spec_ids
 from pyiceberg.schema import Schema, assign_fresh_schema_ids
 from pyiceberg.table import (
@@ -72,7 +74,7 @@
 from pyiceberg.typedef import EMPTY_DICT, UTF8, IcebergBaseModel, Identifier, Properties
 from pyiceberg.types import transform_dict_value_to_str
 from pyiceberg.utils.deprecated import deprecation_message
-from pyiceberg.utils.properties import get_header_properties, property_as_bool
+from pyiceberg.utils.properties import get_first_property_value, get_header_properties, property_as_bool
 
 if TYPE_CHECKING:
     import pyarrow as pa
@@ -390,11 +392,17 @@ class SigV4Adapter(HTTPAdapter):
             def __init__(self, **properties: str):
                 super().__init__()
                 self._properties = properties
+                self._boto_session = boto3.Session(
+                    region_name=get_first_property_value(self._properties, AWS_REGION),
+                    botocore_session=self._properties.get(BOTOCORE_SESSION),
+                    aws_access_key_id=get_first_property_value(self._properties, AWS_ACCESS_KEY_ID),
+                    aws_secret_access_key=get_first_property_value(self._properties, AWS_SECRET_ACCESS_KEY),
+                    aws_session_toxken=get_first_property_value(self._properties, AWS_SESSION_TOKEN),
+                )
 
             def add_headers(self, request: PreparedRequest, **kwargs: Any) -> None:  # pylint: disable=W0613
-                boto_session = boto3.Session()
-                credentials = boto_session.get_credentials().get_frozen_credentials()
-                region = self._properties.get(SIGV4_REGION, boto_session.region_name)
+                credentials = self._boto_session.get_credentials().get_frozen_credentials()
+                region = self._properties.get(SIGV4_REGION, self._boto_session.region_name)
                 service = self._properties.get(SIGV4_SERVICE, "execute-api")
 
                 url = str(request.url).split("?")[0]