update dependencies and fix bug on has_dataset

dlt/destinations/impl/databricks/sql_client.py

@@ -1,6 +1,7 @@
 from contextlib import contextmanager, suppress
 from typing import Any, AnyStr, ClassVar, Iterator, Optional, Sequence, List, Union, Dict
 
+import pendulum
 from databricks import sql as databricks_lib
 from databricks.sql.client import (
     Connection as DatabricksSqlConnection,
@@ -65,6 +66,18 @@ def rollback_transaction(self) -> None:
     def native_connection(self) -> "DatabricksSqlConnection":
         return self._conn
 
+    def has_dataset(self) -> bool:
+        query = """
+    SELECT CATALOG_NAME
+    FROM INFORMATION_SCHEMA.SCHEMATA
+    WHERE """
+        db_params = self.fully_qualified_dataset_name(escape=False).split(".", 2)
+        if len(db_params) == 2:
+            query += " catalog_name = %s AND "
+        query += "schema_name = %s"
+        rows = self.execute_sql(query, *db_params)
+        return len(rows) > 0
+
     def drop_dataset(self) -> None:
         self.execute_sql("DROP SCHEMA IF EXISTS %s CASCADE;" % self.fully_qualified_dataset_name())
 
@@ -89,27 +102,21 @@ def execute_sql(
     def execute_query(self, query: AnyStr, *args: Any, **kwargs: Any) -> Iterator[DBApiCursor]:
         curr: DBApiCursor = None
         # TODO: databricks connector 3.0.0 will use :named paramstyle only
-        # if args:
-        #     keys = [f"arg{i}" for i in range(len(args))]
-        #     # Replace position arguments (%s) with named arguments (:arg0, :arg1, ...)
-        #     # query = query % tuple(f":{key}" for key in keys)
-        #     db_args = {}
-        #     for key, db_arg in zip(keys, args):
-        #         # Databricks connector doesn't accept pendulum objects
-        #         if isinstance(db_arg, pendulum.DateTime):
-        #             db_arg = to_py_datetime(db_arg)
-        #         elif isinstance(db_arg, pendulum.Date):
-        #             db_arg = to_py_date(db_arg)
-        #         db_args[key] = db_arg
-        # else:
-        #     db_args = None
-        db_args: Optional[Union[Dict[str, Any], Sequence[Any]]]
-        if kwargs:
-            db_args = kwargs
-        elif args:
-            db_args = args
+        if args:
+            keys = [f"arg{i}" for i in range(len(args))]
+            # Replace position arguments (%s) with named arguments (:arg0, :arg1, ...)
+            query = query % tuple(f":{key}" for key in keys)
+            db_args = {}
+            for key, db_arg in zip(keys, args):
+                # Databricks connector doesn't accept pendulum objects
+                if isinstance(db_arg, pendulum.DateTime):
+                    db_arg = to_py_datetime(db_arg)
+                elif isinstance(db_arg, pendulum.Date):
+                    db_arg = to_py_date(db_arg)
+                db_args[key] = db_arg
         else:
             db_args = None
+        db_args: Optional[Union[Dict[str, Any], Sequence[Any]]]
         with self._conn.cursor() as curr:
             curr.execute(query, db_args)
             yield DBApiCursorImpl(curr)  # type: ignore[abstract]