Repository: intake/intake
Pull request #866 (2025-03-05 18:04:14)

Add test for cat_mapper


Patch:
diff --git a/intake/readers/transform.py b/intake/readers/transform.py
index 5337d3c..c163f0b 100644
--- a/intake/readers/transform.py
+++ b/intake/readers/transform.py
@@ -111,25 +111,31 @@ class GetItem(BaseConverter):
         return data[item]
 
 
-class CatalogMapper(BaseConverter):
-    instances = {"intake:Catalog": "intake:Catalog"}
+def identity(x):
+    return x
+
 
-    def __init__(self, func, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.func = func
-        self.args = args
-        self.kwargs = kwargs
+class CatalogMapper(BaseConverter):
+    instances = {"intake.readers.entry:Catalog": "intake.readers.entry:Catalog"}
 
     def run(
-        self, in_cat: intake.Catalog, *args, transform=True, name_arg=None, read=False, **kwargs
+        self,
+        in_cat: intake.Catalog,
+        func,
+        *args,
+        transform=True,
+        name_arg=None,
+        read=False,
+        **kwargs,
     ):
         """
         Parameters
         ----------
+        func: function to apply to each entry (as callable or string equivalent)
         transform: do we expect this to be a named transform that intake
             already knows about?
-        name_arg: if give, pass the entry name to the action to be
-            performed using this ad the kwarg name
+        name_arg: if given, pass the entry name to the action to be
+            performed using this as the kwarg name
         read: if True, execute the pipeline produced. This might be used
             where the pipeline output is itself another reader.
         """
@@ -138,9 +144,10 @@ class CatalogMapper(BaseConverter):
             if name_arg:
                 kwargs[name_arg] = name
             if transform:
-                pipe = in_cat[name].__getattr__(self.func)(*args, **kwargs)
+                pipe = in_cat[name].__getattr__(func)(*args, **kwargs)
             else:
-                func = intake.import_name(self.func)
+                if isinstance(func, str):
+                    func = intake.import_name(func)
                 pipe = in_cat[name].apply(func, *args, **kwargs)
             if read:
                 out[name] = pipe.read()


Test patch:
diff --git a/intake/readers/tests/test_workflows.py b/intake/readers/tests/test_workflows.py
index 6459c08..fe83073 100644
--- a/intake/readers/tests/test_workflows.py
+++ b/intake/readers/tests/test_workflows.py
@@ -164,3 +164,22 @@ def test_custom_cache(dataframe_file, tmpdir, df):
     # but condition still picks read from cache
     out = reader2.read()
     assert df.equals(out)
+
+
+def test_cat_mapper(dataframe_file):
+    # setup
+    cat = intake.Catalog()
+    data = intake.readers.datatypes.CSV(url=dataframe_file)
+    cat["one"] = readers.PandasCSV(data)
+    cat["two"] = readers.PandasCSV(data, usecols=[0])
+    cat.to_yaml_file("memory://cat.yaml")
+
+    reader = intake.readers.YAMLCatalogReader("memory://cat.yaml")
+    reader2 = reader.transform.CatalogMapper(getattr, "shape", transform=False)
+
+    cat2 = reader2.read()
+    assert "one" in cat2 and "two" in cat2
+    result1 = cat2["one"].read()
+    assert result1 == (300, 3)
+    result2 = cat2["two"].read()
+    assert result2 == (300, 1)