fix(dask): raise TypeError with informative message in ibis.dask.connect

ibis/backends/dask/__init__.py

@@ -47,6 +47,12 @@ def do_connect(
         # register dispatchers
         from ibis.backends.dask import udf  # noqa: F401
 
+        for k, v in dictionary.items():
+            if not isinstance(v, (dd.DataFrame, pd.DataFrame)):
+                raise TypeError(
+                    f"Expected an instance of 'dask.dataframe.DataFrame' for {k!r},"
+                    f" got an instance of '{type(v).__name__}' instead."
+                )
         super().do_connect(dictionary)
 
     @property

ibis/backends/dask/tests/test_client.py

@@ -1,3 +1,5 @@
+import re
+
 import numpy as np
 import pandas as pd
 import pytest
@@ -107,3 +109,30 @@ def test_datetime64_infer(client, unit):
     expr = ibis.literal(value, type='timestamp')
     result = client.execute(expr)
     assert result == value
+
+
+def test_invalid_connection_parameter_types(npartitions):
+    # Check that the user receives a TypeError with an informative message when
+    # passing invalid an connection parameter to the backend.
+    expected_msg = re.escape(
+        "Expected an instance of 'dask.dataframe.DataFrame' for 'invalid_str',"
+        " got an instance of 'str' instead."
+    )
+    with pytest.raises(TypeError, match=expected_msg):
+        ibis.dask.connect(
+            {
+                "valid_dask_df": dd.from_pandas(
+                    pd.DataFrame({'a': [1, 2, 3], 'b': list('abc')}),
+                    npartitions=npartitions,
+                ),
+                "valid_pandas_df": pd.DataFrame({'a': [1, 2, 3], 'b': list('abc')}),
+                "invalid_str": "file.csv",
+            }
+        )
+
+    expeced_msg = re.escape(
+        "Expected an instance of 'dask.dataframe.DataFrame' for 'df', "
+        "got an instance of 'str' instead."
+    )
+    with pytest.raises(TypeError, match=expeced_msg):
+        ibis.dask.from_dataframe("file.csv")