storage: index as a dir if no glob (#108)

* storage: index as a dir if no glob

* add tests for enlist source on globs, dirs, file paths

Author: shcheklein

src/datachain/catalog/catalog.py

@@ -1,4 +1,5 @@
 import ast
+import glob
 import io
 import json
 import logging
@@ -709,7 +710,12 @@ def enlist_source(
 
         client_config = client_config or self.client_config
         client, path = self.parse_url(source, **client_config)
-        prefix = posixpath.dirname(path)
+        stem = os.path.basename(os.path.normpath(path))
+        prefix = (
+            posixpath.dirname(path)
+            if glob.has_magic(stem) or client.fs.isfile(source)
+            else path
+        )
         storage_dataset_name = Storage.dataset_name(
             client.uri, posixpath.join(prefix, "")
         )

tests/func/test_catalog.py

@@ -1068,6 +1068,45 @@ def test_storage_stats(cloud_test_catalog):
     assert stats.size == 15
 
 
+@pytest.mark.parametrize("cloud_type", ["s3", "azure", "gs"], indirect=True)
+def test_enlist_source_handles_slash(cloud_test_catalog):
+    catalog = cloud_test_catalog.catalog
+    src_uri = cloud_test_catalog.src_uri
+
+    catalog.enlist_source(f"{src_uri}/dogs", ttl=1234)
+    stats = catalog.storage_stats(src_uri)
+    assert stats.num_objects == len(DEFAULT_TREE["dogs"])
+    assert stats.size == 15
+
+    catalog.enlist_source(f"{src_uri}/dogs/", ttl=1234, force_update=True)
+    stats = catalog.storage_stats(src_uri)
+    assert stats.num_objects == len(DEFAULT_TREE["dogs"])
+    assert stats.size == 15
+
+
+@pytest.mark.parametrize("cloud_type", ["s3", "azure", "gs"], indirect=True)
+def test_enlist_source_handles_glob(cloud_test_catalog):
+    catalog = cloud_test_catalog.catalog
+    src_uri = cloud_test_catalog.src_uri
+
+    catalog.enlist_source(f"{src_uri}/dogs/*.jpg", ttl=1234)
+    stats = catalog.storage_stats(src_uri)
+
+    assert stats.num_objects == len(DEFAULT_TREE["dogs"])
+    assert stats.size == 15
+
+
+@pytest.mark.parametrize("cloud_type", ["s3", "azure", "gs"], indirect=True)
+def test_enlist_source_handles_file(cloud_test_catalog):
+    catalog = cloud_test_catalog.catalog
+    src_uri = cloud_test_catalog.src_uri
+
+    catalog.enlist_source(f"{src_uri}/dogs/dog1", ttl=1234)
+    stats = catalog.storage_stats(src_uri)
+    assert stats.num_objects == len(DEFAULT_TREE["dogs"])
+    assert stats.size == 15
+
+
 @pytest.mark.parametrize("from_cli", [False, True])
 def test_garbage_collect(cloud_test_catalog, from_cli, capsys):
     catalog = cloud_test_catalog.catalog

tests/unit/lib/test_datachain.py

@@ -660,7 +660,7 @@ def test_parse_tabular_partitions(tmp_dir, catalog):
 
 def test_parse_tabular_empty(tmp_dir, catalog):
     path = tmp_dir / "test.parquet"
-    with pytest.raises(DataChainParamsError):
+    with pytest.raises(FileNotFoundError):
         DataChain.from_storage(path.as_uri()).parse_tabular()