def _create_session()

in pyiceberg/catalog/rest.py [0:0]

• 20 lines of code
• 13 McCabe index (conditional complexity)

    def _create_session(self) -> Session:
        """Create a request session with provided catalog configuration."""
        session = Session()

        # Sets the client side and server side SSL cert verification, if provided as properties.
        if ssl_config := self.properties.get(SSL):
            if ssl_ca_bundle := ssl_config.get(CA_BUNDLE):
                session.verify = ssl_ca_bundle
            if ssl_client := ssl_config.get(CLIENT):
                if all(k in ssl_client for k in (CERT, KEY)):
                    session.cert = (ssl_client[CERT], ssl_client[KEY])
                elif ssl_client_cert := ssl_client.get(CERT):
                    session.cert = ssl_client_cert

        self._refresh_token(session, self.properties.get(TOKEN))

        # Set HTTP headers
        self._config_headers(session)

        # Configure SigV4 Request Signing
        if property_as_bool(self.properties, SIGV4, False):
            self._init_sigv4(session)

        # Mount custom adapters
        if session_adapters := self.properties.get("session_adapters"):
            for prefix, adapter in session_adapters.items():  # type: ignore
                session.mount(prefix, adapter)
        # Add custom auth
        if session_auth := self.properties.get("session_auth"):
            session.auth = session_auth  # type: ignore

        return session