def _initialize_s3_fs()

in pyiceberg/io/pyarrow.py [0:0]

• 34 lines of code
• 10 McCabe index (conditional complexity)

    def _initialize_s3_fs(self, netloc: Optional[str]) -> FileSystem:
        from pyarrow.fs import S3FileSystem, resolve_s3_region

        # Resolve region from netloc(bucket), fallback to user-provided region
        provided_region = get_first_property_value(self.properties, S3_REGION, AWS_REGION)

        try:
            bucket_region = resolve_s3_region(bucket=netloc)
        except (OSError, TypeError):
            bucket_region = None
            logger.warning(f"Unable to resolve region for bucket {netloc}, using default region {provided_region}")

        bucket_region = bucket_region or provided_region
        if bucket_region != provided_region:
            logger.warning(
                f"PyArrow FileIO overriding S3 bucket region for bucket {netloc}: "
                f"provided region {provided_region}, actual region {bucket_region}"
            )

        client_kwargs: Dict[str, Any] = {
            "endpoint_override": self.properties.get(S3_ENDPOINT),
            "access_key": get_first_property_value(self.properties, S3_ACCESS_KEY_ID, AWS_ACCESS_KEY_ID),
            "secret_key": get_first_property_value(self.properties, S3_SECRET_ACCESS_KEY, AWS_SECRET_ACCESS_KEY),
            "session_token": get_first_property_value(self.properties, S3_SESSION_TOKEN, AWS_SESSION_TOKEN),
            "region": bucket_region,
        }

        if proxy_uri := self.properties.get(S3_PROXY_URI):
            client_kwargs["proxy_options"] = proxy_uri

        if connect_timeout := self.properties.get(S3_CONNECT_TIMEOUT):
            client_kwargs["connect_timeout"] = float(connect_timeout)

        if request_timeout := self.properties.get(S3_REQUEST_TIMEOUT):
            client_kwargs["request_timeout"] = float(request_timeout)

        if role_arn := get_first_property_value(self.properties, S3_ROLE_ARN, AWS_ROLE_ARN):
            client_kwargs["role_arn"] = role_arn

        if session_name := get_first_property_value(self.properties, S3_ROLE_SESSION_NAME, AWS_ROLE_SESSION_NAME):
            client_kwargs["session_name"] = session_name

        if force_virtual_addressing := self.properties.get(S3_FORCE_VIRTUAL_ADDRESSING):
            client_kwargs["force_virtual_addressing"] = property_as_bool(self.properties, force_virtual_addressing, False)

        return S3FileSystem(**client_kwargs)