feat: support SparkJDBCDataset JDBC URLs in credentials

kedro-datasets/RELEASE.md

@@ -14,6 +14,7 @@
 - Fixed `MLRunModel` so user-supplied `load_args` are now passed to `joblib.load()` (previously silently dropped). Added a deserialization warning to the docstring.
 - Hardened `TensorFlowModelDataset`: `safe_mode=True` is now the default for `load_model()` to prevent arbitrary code execution from untrusted model files. Fixed a bug where `tf_device` was lost from `load_args` after the first load call.
 - Added deserialization risk warnings to docstrings of datasets that can execute arbitrary code when loading untrusted files.
+- Added support for supplying `SparkJDBCDataset` JDBC URLs through credentials.
 
 ## Community contributions
 - [samiat4911](https://github.com/samiat4911)

kedro-datasets/kedro_datasets/spark/spark_jdbc_dataset.py

@@ -23,14 +23,19 @@ class SparkJDBCDataset(AbstractDataset[DataFrame, DataFrame]):
         weather:
           type: spark.SparkJDBCDataset
           table: weather_table
-          url: jdbc:postgresql://localhost/test
           credentials: db_credentials
           load_args:
             properties:
               driver: org.postgresql.Driver
           save_args:
             properties:
               driver: org.postgresql.Driver
+
+        # credentials.yml
+        db_credentials:
+          url: jdbc:postgresql://localhost/test
+          user: scott
+          password: tiger
         ```
 
         Using the [Python API](https://docs.kedro.org/en/stable/catalog-data/advanced_data_catalog_usage/):
@@ -67,8 +72,8 @@ class SparkJDBCDataset(AbstractDataset[DataFrame, DataFrame]):
     def __init__(  # noqa: PLR0913
         self,
         *,
-        url: str,
         table: str,
+        url: str | None = None,
         credentials: dict[str, Any] | None = None,
         load_args: dict[str, Any] | None = None,
         save_args: dict[str, Any] | None = None,
@@ -77,7 +82,8 @@ def __init__(  # noqa: PLR0913
         """Creates a new ``SparkJDBCDataset``.
 
         Args:
-            url: A JDBC URL of the form ``jdbc:subprotocol:subname``.
+            url: A JDBC URL of the form ``jdbc:subprotocol:subname``. When not
+                provided, the URL can be supplied as ``url`` in ``credentials``.
             table: The name of the table to load or save data to.
             credentials: A dictionary of JDBC database connection arguments.
                 Normally at least properties ``user`` and ``password`` with
@@ -100,6 +106,9 @@ def __init__(  # noqa: PLR0913
                 when a property is provided with a None value.
         """
 
+        credentials = credentials or {}
+        url = url or credentials.get("url")
+
         if not url:
             raise DatasetError(
                 "'url' argument cannot be empty. Please "
@@ -124,9 +133,14 @@ def __init__(  # noqa: PLR0913
         self._save_args = {**self.DEFAULT_SAVE_ARGS, **(save_args or {})}
 
         # Update properties in load_args and save_args with credentials.
-        if credentials is not None:
+        credentials_properties = {
+            cred_key: cred_value
+            for cred_key, cred_value in credentials.items()
+            if cred_key != "url"
+        }
+        if credentials_properties:
             # Check credentials for bad inputs.
-            for cred_key, cred_value in credentials.items():
+            for cred_key, cred_value in credentials_properties.items():
                 if cred_value is None:
                     raise DatasetError(
                         f"Credential property '{cred_key}' cannot be None. "
@@ -135,8 +149,14 @@ def __init__(  # noqa: PLR0913
 
             load_properties = self._load_args.get("properties", {})
             save_properties = self._save_args.get("properties", {})
-            self._load_args["properties"] = {**load_properties, **credentials}
-            self._save_args["properties"] = {**save_properties, **credentials}
+            self._load_args["properties"] = {
+                **load_properties,
+                **credentials_properties,
+            }
+            self._save_args["properties"] = {
+                **save_properties,
+                **credentials_properties,
+            }
 
     def _describe(self) -> dict[str, Any]:
         load_args = self._load_args

kedro-datasets/tests/spark/test_spark_jdbc_dataset.py

@@ -39,7 +39,19 @@ def test_missing_url():
         " URL of the form 'jdbc:subprotocol:subname'."
     )
     with pytest.raises(DatasetError, match=error_message):
-        SparkJDBCDataset(url=None, table="dummy_table")
+        SparkJDBCDataset(table="dummy_table")
+
+
+def test_missing_url_when_credentials_do_not_contain_url():
+    error_message = (
+        "'url' argument cannot be empty. Please provide a JDBC"
+        " URL of the form 'jdbc:subprotocol:subname'."
+    )
+    with pytest.raises(DatasetError, match=error_message):
+        SparkJDBCDataset(
+            table="dummy_table",
+            credentials={"user": "dummy_user", "password": "dummy_pw"},
+        )
 
 
 def test_missing_table():
@@ -69,6 +81,50 @@ def test_save_credentials(mocker, spark_jdbc_args_credentials):
     )
 
 
+def test_save_credentials_url(mocker):
+    mock_data = mocker.Mock()
+    credentials = {
+        "url": "credentials_url",
+        "user": "dummy_user",
+        "password": "dummy_pw",
+    }
+    dataset = SparkJDBCDataset(table="dummy_table", credentials=credentials)
+
+    dataset.save(mock_data)
+
+    mock_data.write.jdbc.assert_called_with(
+        "credentials_url",
+        "dummy_table",
+        properties={"user": "dummy_user", "password": "dummy_pw"},
+    )
+    assert credentials == {
+        "url": "credentials_url",
+        "user": "dummy_user",
+        "password": "dummy_pw",
+    }
+
+
+def test_save_explicit_url_takes_precedence_over_credentials_url(mocker):
+    mock_data = mocker.Mock()
+    dataset = SparkJDBCDataset(
+        url="dummy_url",
+        table="dummy_table",
+        credentials={
+            "url": "credentials_url",
+            "user": "dummy_user",
+            "password": "dummy_pw",
+        },
+    )
+
+    dataset.save(mock_data)
+
+    mock_data.write.jdbc.assert_called_with(
+        "dummy_url",
+        "dummy_table",
+        properties={"user": "dummy_user", "password": "dummy_pw"},
+    )
+
+
 def test_save_args(mocker, spark_jdbc_args_save_load):
     mock_data = mocker.Mock()
     dataset = SparkJDBCDataset(**spark_jdbc_args_save_load)
@@ -108,6 +164,31 @@ def test_load_credentials(mocker, spark_jdbc_args_credentials):
     )
 
 
+def test_load_credentials_url(mocker):
+    spark = mocker.patch(
+        "kedro_datasets.spark.spark_jdbc_dataset.get_spark"
+    ).return_value
+    credentials = {
+        "url": "credentials_url",
+        "user": "dummy_user",
+        "password": "dummy_pw",
+    }
+    dataset = SparkJDBCDataset(table="dummy_table", credentials=credentials)
+
+    dataset.load()
+
+    spark.read.jdbc.assert_called_with(
+        "credentials_url",
+        "dummy_table",
+        properties={"user": "dummy_user", "password": "dummy_pw"},
+    )
+    assert credentials == {
+        "url": "credentials_url",
+        "user": "dummy_user",
+        "password": "dummy_pw",
+    }
+
+
 def test_load_args(mocker, spark_jdbc_args_save_load):
     spark = mocker.patch(
         "kedro_datasets.spark.spark_jdbc_dataset.get_spark"