decouple session class from existing Connection

ensure maintenance of current APIs of Connection while delegating
responsibility

Signed-off-by: varun-edachali-dbx <varun.edachali@databricks.com>

Author: varun-edachali-dbx

src/databricks/sql/client.py

@@ -19,6 +19,8 @@
     OperationalError,
     SessionAlreadyClosedError,
     CursorAlreadyClosedError,
+    Error,
+    NotSupportedError,
 )
 from databricks.sql.thrift_api.TCLIService import ttypes
 from databricks.sql.backend.thrift_backend import ThriftDatabricksClient
@@ -249,6 +251,20 @@ def read(self) -> Optional[OAuthToken]:
             + str(self.get_session_id_hex())
         )
 
+        # Create the session
+        self.session = Session(
+            server_hostname,
+            http_path,
+            http_headers,
+            session_configuration,
+            catalog,
+            schema,
+            _use_arrow_native_complex_types,
+            **kwargs
+        )
+        
+        logger.info("Successfully opened connection with session " + str(self.get_session_id_hex()))
+
         self.use_inline_params = self._set_use_inline_params_with_warning(
             kwargs.get("use_inline_params", False)
         )
@@ -294,7 +310,7 @@ def __exit__(self, exc_type, exc_value, traceback):
         return False
 
     def __del__(self):
-        if self.open:
+        if self.session.open:
             logger.debug(
                 "Closing unclosed connection for session "
                 "{}".format(self.get_session_id_hex())
@@ -328,13 +344,6 @@ def get_protocol_version(openSessionResp):
         """Delegate to Session class static method"""
         return Session.get_protocol_version(openSessionResp)
 
-    @property
-    def open(self) -> bool:
-        """Return whether the connection is open by checking if the session is open."""
-        # NOTE: we have to check for the existence of session in case the __del__ is called
-        # before the session is instantiated
-        return hasattr(self, "session") and self.session.open
-
     def cursor(
         self,
         arraysize: int = DEFAULT_ARRAY_SIZE,
@@ -345,7 +354,7 @@ def cursor(
 
         Will throw an Error if the connection has been closed.
         """
-        if not self.open:
+        if not self.session.open:
             raise Error("Cannot create cursor from closed connection")
 
         cursor = Cursor(
@@ -1490,7 +1499,7 @@ def close(self) -> None:
             if (
                 self.op_state != ttypes.TOperationState.CLOSED_STATE
                 and not self.has_been_closed_server_side
-                and self.connection.open
+                and self.connection.session.open
             ):
                 self.backend.close_command(self.command_id)
         except RequestError as e: