Repository: intake/intake
Pull request #630 (2021-11-24 15:09:03)

Allow user parameters throughout catalog
Currently, catalog-level parameters are established as an argument `user_parameters=` to `Catalog`, or a block in the YAML file:
```
metadata:
  parameters:
    var_name:
      type: str
      description: description
      default: something
```



(requires documentation and tests)


Patch:
diff --git a/docs/source/catalog.rst b/docs/source/catalog.rst
index ac0eae4..0d31d18 100644
--- a/docs/source/catalog.rst
+++ b/docs/source/catalog.rst
@@ -293,6 +293,9 @@ resolve to ``"http://server:port/user/"``
 Parameter Definition
 --------------------
 
+Source parameters
+'''''''''''''''''
+
 A source definition can contain a "parameters" block.
 Expressed in YAML, a parameter may look as follows:
 
@@ -329,6 +332,55 @@ can have one of two uses:
 Note: the ``datetime`` type accepts multiple values:
 Python datetime, ISO8601 string,  Unix timestamp int, "now" and  "today".
 
+Catalog parameters
+''''''''''''''''''
+
+You can also define user parameters at the catalog level. This applies the parameter to
+all entries within that catalog, without having to define it for each and every entry.
+Furthermore, catalogs dested within the catalog will also inherit the parameter(s).
+
+For example, with the following spec
+
+.. code-block:: yaml
+
+    metadata:
+      version: 1
+      parameters:
+        bucket:
+          type: str
+          description: description
+          default: test_bucket
+    sources:
+      param_source:
+        driver: parquet
+        description: description
+        args:
+          urlpath: s3://{{bucket}}/file.parquet
+      subcat:
+        driver: yaml_file
+        path: "{{CATALOG_DIR}}/other.yaml"
+
+If ``cat`` is the corresponsing catalog instance,
+the URL of source ``cat.param_source`` will evaluate to "s3://test_bucket/file.parquet" by default, but
+the parameter can be overridden with ``cat.param_source(bucket="other_bucket")``. Also, any
+entries of ``subcat``, another catalog referenced from here, would also have the "bucket"-named
+parameter attached to all sources. Of course, those sources do no need to make use of the
+parameter.
+
+To change the default, we can gerenate a new instance
+
+.. code-block:: python
+
+    cat2 = cat(bucket="production")  # sets default value of "bucket" for cat2
+    subcat = cat.subcat(bucket="production")  # sets default only for the nested catalog
+
+Of course, in these situations you can still override the value of the parameter for any
+source, or pass explicit values for the arguments of the source, as normal.
+
+For cases where the catalog is not defined in a YAML spec, the argument ``user_parameters``
+to the constructor takes the same form as ``parameters`` above: a dict of user parameters,
+either as ``UserParameter`` instances or as a dictionary spec for each one.
+
 Driver Selection
 ----------------
 
diff --git a/intake/catalog/base.py b/intake/catalog/base.py
index 8061837..bb53f54 100644
--- a/intake/catalog/base.py
+++ b/intake/catalog/base.py
@@ -42,7 +42,7 @@ class Catalog(DataSource):
 
     def __init__(self, entries=None, name=None, description=None, metadata=None,
                  ttl=60, getenv=True, getshell=True,
-                 persist_mode='default', storage_options=None):
+                 persist_mode='default', storage_options=None, user_parameters=None):
         """
         Parameters
         ----------
@@ -83,6 +83,15 @@ class Catalog(DataSource):
         self.getenv = getenv
         self.getshell = getshell
         self.storage_options = storage_options
+        if isinstance(user_parameters, dict) and user_parameters:
+            from .local import UserParameter
+            self.user_parameters = {name: (UserParameter(name=name, **up) if isinstance(up, dict)
+                                           else up)
+                                    for name, up in user_parameters.items()}
+        elif isinstance(user_parameters, (list, tuple)):
+            self.user_parameters = {up["name"]: up for up in user_parameters}
+        else:
+            self.user_parameters = {}
         if persist_mode not in ['always', 'never', 'default']:
             # should be True, False, None ?
             raise ValueError('Persist mode (%s) not understood' % persist_mode)
@@ -181,7 +190,9 @@ class Catalog(DataSource):
             getenv=self.getenv,
             getshell=self.getshell,
             metadata=(self.metadata or {}).copy(),
-            storage_options=self.storage_options)
+            storage_options=self.storage_options,
+            user_parameters=self.user_parameters.copy()
+        )
         cat.metadata['search'] = {'text': text, 'upstream': self.name}
         cat.cat = self
         for e in entries.values():
@@ -304,8 +315,38 @@ class Catalog(DataSource):
         entry = self._entries[name]
         entry._catalog = self
         entry._pmode = self.pmode
+
+        up_names = set((up["name"] if isinstance(up, dict) else up.name)
+                        for up in entry._user_parameters)
+        ups = [up for name, up in self.user_parameters.items() if name not in up_names]
+        entry._user_parameters = ups + (entry._user_parameters or [])
         return entry()
 
+    def configure_new(self, **kwargs):
+        from .local import UserParameter
+        ups = {}
+        for k, v in kwargs.copy().items():
+            for up in self.user_parameters.values():
+                if isinstance(up, dict):
+                    if k == up["name"]:
+                        kw = up.copy()
+                        kw['default'] = v
+                        ups[k] = UserParameter(**kw)
+                        kwargs.pop(k)
+                else:
+                    if k == up.name:
+                        kw = up._captured_init_kwargs.copy()
+                        kw['default'] = v
+                        kw["name"] = k
+                        ups[k] = UserParameter(**kw)
+                        kwargs.pop(k)
+
+        new = super().configure_new(**kwargs)
+        new.user_parameters.update(ups)
+        return new
+
+    __call__ = get = configure_new
+
     @reload_on_change
     def _get_entries(self):
         return self._entries
@@ -390,14 +431,14 @@ class Catalog(DataSource):
         nested directories with cat.name1.name2, cat['name1.name2'] *or*
         cat['name1', 'name2']
         """
-        if not isinstance(key, list) and key in self._get_entries():
+        if not isinstance(key, list) and key in self:
             # triggers reload_on_change
-            e = self._entries[key]
-            e._catalog = self
-            e._pmode = self.pmode
-            if e.container == 'catalog':
-                return e(name=key)
-            return e()
+            s = self._get_entry(key)
+            if s.container == 'catalog':
+                s.name = key
+                s.user_parameters.update(self.user_parameters.copy())
+                return s
+            return s
         if isinstance(key, str) and '.' in key:
             key = key.split('.')
         if isinstance(key, list):
diff --git a/intake/catalog/local.py b/intake/catalog/local.py
index ab9005e..e89c193 100644
--- a/intake/catalog/local.py
+++ b/intake/catalog/local.py
@@ -691,28 +691,14 @@ class YAMLFileCatalog(Catalog):
         cfg = result.data
 
         self._entries = {}
-        try:
-            shared_parameters = data["metadata"]["parameters"]
-            params = [
-                UserParameter(name, **attrs)
-                for name, attrs in shared_parameters.items()
-            ]
-        except KeyError:
-            params = None
+        shared_parameters = data.get("metadata", {}).get("parameters", {})
+        self.user_parameters.update({
+            name: UserParameter(name, **attrs)
+            for name, attrs in shared_parameters.items()
+        })
 
         for entry in cfg['data_sources']:
             entry._catalog = self
-            if params is not None:
-                try:
-                    # Note that putting the entry parameters after the global parameters
-                    # means that local parameters will overwrite any inherited parameters
-                    all_params = params + entry._user_parameters
-                except AttributeError:
-                    all_params = params
-
-                entry.__setattr__("_user_parameters", all_params)
-                # Necessary for a copy of the entry to have the same parameters
-                entry._captured_init_kwargs["parameters"] = all_params
             self._entries[entry.name] = entry
             entry._filesystem = self.filesystem
 
@@ -793,11 +779,14 @@ class YAMLFilesCatalog(Catalog):
             if f.path not in self._cats:
                 entry = LocalCatalogEntry(name, "YAML file: %s" % name,
                                           'yaml_file_cat', True,
-                                          kwargs, [], {}, self.metadata, d)
+                                          kwargs, [], [], self.metadata, d)
                 if self._flatten:
                     # store a concrete Catalog
                     try:
-                        self._cats[f.path] = entry()
+                        cat = entry()
+                        cat.reload()
+                        self.user_parameters.update(cat.user_parameters)
+                        self._cats[f.path] = cat
                     except IOError as e:
                         logger.info('Loading "%s" as a catalog failed: %s'
                                     '' % (entry, e))
@@ -817,10 +806,6 @@ class YAMLFilesCatalog(Catalog):
                 entries.update(entry._entries)
             else:
                 entries[entry._name] = entry
-        # This is thread safe as long as the hash values for entries are
-        # computed in C --- i.e. if they are built-in types like strings. If
-        # they implement __hash__ in Python then this is not threadsafe, and it
-        # would need a lock.
         self._entries.update(entries)
 
 
diff --git a/intake/catalog/remote.py b/intake/catalog/remote.py
index 9406c46..7af9df2 100644
--- a/intake/catalog/remote.py
+++ b/intake/catalog/remote.py
@@ -156,7 +156,6 @@ class RemoteCatalog(Catalog):
                 auth=self.auth,
                 http_args=self.http_args,
                 page_size=self._page_size,
-                persist_mode=self.pmode,
                 # user_parameters=user_parameters,
                 **source)
         return page
@@ -182,7 +181,6 @@ class RemoteCatalog(Catalog):
             auth=self.auth,
             http_args=self.http_args,
             page_size=self._page_size,
-            persist_mode=self.pmode,
             **info['source'])
 
     def _get_http_args(self, params):
@@ -276,7 +274,6 @@ class RemoteCatalog(Catalog):
             url=self.url,
             http_args=self.http_args,
             source_id=source_id,
-            persist_mode=self.pmode,
             name="")
         cat.cat = self
         return cat
@@ -392,8 +389,8 @@ class RemoteCatalogEntry(CatalogEntry):
     """An entry referring to a remote data definition"""
     def __init__(self, url, auth, name=None, user_parameters=None,
                  container=None, description='', metadata=None,
-                 http_args=None, page_size=None, persist_mode="default", direct_access=False,
-                 getenv=True, getshell=True, **kwargs):
+                 http_args=None, page_size=None, direct_access=False,
+                 getenv=True, getshell=True, **kwags):
         """
 
         Parameters
@@ -415,8 +412,6 @@ class RemoteCatalogEntry(CatalogEntry):
         self.description = description
         self._metadata = metadata or {}
         self._page_size = page_size
-        # Persist mode describing a nested RemoteCatalog
-        self.catalog_pmode = persist_mode
         self._user_parameters = [remake_instance(up)
                                  if (isinstance(up, dict) and 'cls' in up)
                                  else up
@@ -428,8 +423,6 @@ class RemoteCatalogEntry(CatalogEntry):
         
         super(RemoteCatalogEntry, self).__init__(getenv=getenv,
                                                  getshell=getshell)
-
-        # Persist mode for the RemoteCatalogEntry
         self._pmode = "never"
 
     def describe(self):
@@ -463,12 +456,11 @@ class RemoteCatalogEntry(CatalogEntry):
             page_size=self._page_size,
             auth=self.auth,
             getenv=self.getenv,
-            persist_mode=self.catalog_pmode,
             getshell=self.getshell)
 
 
 def open_remote(url, entry, container, user_parameters, description, http_args,
-                page_size=None, persist_mode=None, auth=None, getenv=None, getshell=None):
+                page_size=None, auth=None, getenv=None, getshell=None):
     """Create either local direct data source or remote streamed source"""
     from intake.container import container_map
     import msgpack
@@ -507,14 +499,13 @@ def open_remote(url, entry, container, user_parameters, description, http_args,
                 response.update({'auth': auth,
                                  'getenv': getenv,
                                  'getshell': getshell,
-                                 'page_size': page_size,
-                                 'persist_mode': persist_mode
+                                 'page_size': page_size
                                  # TODO ttl?
                                  # TODO storage_options?
                                  })
             source = container_map[container](url, http_args, **response)
         source.description = description
         return source
+
     else:
         raise Exception('Server error: %d, %s' % (req.status_code, req.reason))
-


Test patch:
diff --git a/intake/catalog/tests/test_local.py b/intake/catalog/tests/test_local.py
index c1f6304..b17bfc8 100644
--- a/intake/catalog/tests/test_local.py
+++ b/intake/catalog/tests/test_local.py
@@ -803,23 +803,6 @@ def test_cat_dictlike(catalog1):
     assert list(catalog1.items()) == list(zip(catalog1.keys(), catalog1.values()))
 
 
-@pytest.fixture
-def inherit_params_cat():
-    with tempfile.TemporaryDirectory() as tmp_dir:
-        tmp_path = posixpath.join(tmp_dir, "intake")
-        target_catalog = copy_test_file("catalog_inherit_params.yml", tmp_path)
-        return open_catalog(target_catalog)
-
-
-@pytest.fixture
-def inherit_params_multiple_cats():
-    with tempfile.TemporaryDirectory() as tmp_dir:
-        tmp_path = posixpath.join(tmp_dir, "intake")
-        copy_test_file("catalog_inherit_params.yml", tmp_path)
-        copy_test_file("catalog_nested_sub.yml", tmp_path)
-        return open_catalog(tmp_path + "/*.yml")
-
-
 def test_inherit_params(inherit_params_cat):
     assert inherit_params_cat.param._urlpath == "s3://test_bucket/file.parquet"
 
diff --git a/intake/catalog/tests/test_parameters.py b/intake/catalog/tests/test_parameters.py
index 2e915d3..a874f42 100644
--- a/intake/catalog/tests/test_parameters.py
+++ b/intake/catalog/tests/test_parameters.py
@@ -1,5 +1,6 @@
 import os
 import pytest
+import intake
 from intake.catalog.local import LocalCatalogEntry, UserParameter
 from intake.source.base import DataSource
 
@@ -217,3 +218,20 @@ def test_unknown():
                           parameters=[up])
     s = e()
     assert s.kwargs['arg1'] == ""
+
+
+def test_catalog_passthrough():
+    root = os.path.dirname(os.path.dirname(os.path.dirname(__file__)))
+    cat = intake.open_catalog(os.path.join(root, "tests/catalog_inherit_params.yml"))
+    assert set(cat.subcat.user_parameters) == {"bucket", "inner"}
+    url = cat.subcat.ex2.urlpath
+    assert url == "test_bucket/test_name"
+    url = cat.subcat.ex2(bucket="hi", inner="ho").urlpath
+    assert url == "hi/ho"
+
+    # test clone
+    cat2 = cat(bucket="yet")
+    url = cat2.subcat(inner="another").ex2.urlpath
+    assert url == "yet/another"
+    url = cat2.subcat(inner="another").ex2(bucket="hi").urlpath
+    assert url == "hi/another"
diff --git a/intake/catalog/tests/test_reload_integration.py b/intake/catalog/tests/test_reload_integration.py
index 4af1d5f..0b78e8b 100644
--- a/intake/catalog/tests/test_reload_integration.py
+++ b/intake/catalog/tests/test_reload_integration.py
@@ -45,9 +45,9 @@ sources:
     description: example1 source plugin
     driver: example1
     args: {}
-        ''')
+''')
 
-    time.sleep(2)
+    time.sleep(1)
 
     yield intake_server
     os.remove(fullname)
@@ -59,22 +59,13 @@ def test_reload_updated_config(intake_server_with_config):
     entries = list(catalog)
     assert entries == ['use_example1']
 
-    with open(os.path.join(TMP_DIR, YAML_FILENAME), 'w') as f:
+    with open(os.path.join(TMP_DIR, YAML_FILENAME), 'a') as f:
         f.write('''
-plugins:
-  source:
-    - module: intake.catalog.tests.example1_source
-    - module: intake.catalog.tests.example_plugin_dir.example2_source
-sources:
-  use_example1:
-    description: example1 source plugin
-    driver: example1
-    args: {}
   use_example1_1:
     description: example1 other
     driver: example1
     args: {}
-        ''')
+''')
 
     time.sleep(1.2)
 
diff --git a/intake/conftest.py b/intake/conftest.py
index d179bd3..dd3fb07 100644
--- a/intake/conftest.py
+++ b/intake/conftest.py
@@ -9,15 +9,18 @@ import os
 import subprocess
 import time
 
+import posixpath
 import pytest
 import requests
+import tempfile
 
-from intake import config
+from intake import config, open_catalog
 from intake.container import persist
 from intake.util_tests import ex, PY2
 from intake.utils import make_path_posix
 from intake.source.base import DataSource, Schema
 from intake import register_driver
+from intake.tests.test_utils import copy_test_file
 
 here = os.path.dirname(__file__)
 
@@ -201,3 +204,29 @@ def env(temp_cache, tempdir):
     env["INTAKE_CONF_DIR"] = intake.config.confdir
     env['INTAKE_CACHE_DIR'] = intake.config.conf['cache_dir']
     return env
+
+
+@pytest.fixture
+def inherit_params_cat():
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        tmp_path = posixpath.join(tmp_dir, "intake")
+        target_catalog = copy_test_file("catalog_inherit_params.yml", tmp_path)
+        return open_catalog(target_catalog)
+
+
+@pytest.fixture
+def inherit_params_multiple_cats():
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        tmp_path = posixpath.join(tmp_dir, "intake")
+        copy_test_file("catalog_inherit_params.yml", tmp_path)
+        copy_test_file("catalog_nested_sub.yml", tmp_path)
+        return open_catalog(tmp_path + "/*.yml")
+
+
+@pytest.fixture
+def inherit_params_subcat():
+    with tempfile.TemporaryDirectory() as tmp_dir:
+        tmp_path = posixpath.join(tmp_dir, "intake")
+        target_catalog = copy_test_file("catalog_inherit_params.yml", tmp_path)
+        copy_test_file("catalog_nested_sub.yml", tmp_path)
+        return open_catalog(target_catalog)
diff --git a/intake/tests/catalog_inherit_params.yml b/intake/tests/catalog_inherit_params.yml
index 3ae384e..c520762 100644
--- a/intake/tests/catalog_inherit_params.yml
+++ b/intake/tests/catalog_inherit_params.yml
@@ -32,3 +32,12 @@ sources:
         default: local_filename.parquet
     args:
       urlpath: s3://{{bucket}}/{{filename}}
+  subcat:
+    driver: yaml_file_cat
+    args:
+      path: "{{CATALOG_DIR}}/catalog_nested_sub.yml"
+      user_parameters:
+        inner:
+          type: str
+          description: description
+          default: test_name
diff --git a/intake/tests/catalog_nested_sub.yml b/intake/tests/catalog_nested_sub.yml
index 9759ee3..9fcb945 100644
--- a/intake/tests/catalog_nested_sub.yml
+++ b/intake/tests/catalog_nested_sub.yml
@@ -4,3 +4,8 @@ sources:
     driver: csv
     args:
       urlpath: ""
+  ex2:
+    description: this is a sub-resource
+    driver: csv
+    args:
+      urlpath: "{{bucket}}/{{inner}}"