Repository: intake/intake
Pull request #528 (2020-08-28 14:52:31)

Refactor DataSource
Make simplified DataSourceBase, and mixins containing the various optional behaviours which were all within DataSource.

For now, DataSource will depend on all the mixins, so existing behaviour will not change.


Patch:
diff --git a/.travis.yml b/.travis.yml
index 2a472d4..be34d4b 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -26,7 +26,7 @@ jobs:
       install:
         - sudo apt update
         - sudo apt install libsnappy-dev
-        - pip install panel bokeh dask pytest zarr
+        - pip install panel bokeh dask[complete] pytest zarr aiohttp requests
         - pip install -e .[complete]
       script:
         - pytest
@@ -36,7 +36,7 @@ jobs:
       install:
         - sudo apt update
         - sudo apt install libsnappy-dev
-        - pip install panel bokeh dask pytest zarr
+        - pip install panel bokeh dask[complete] pytest zarr aiohttp requests
         - pip install -e .[complete]
       script:
         - pytest
diff --git a/conda/meta.yaml b/conda/meta.yaml
deleted file mode 100644
index 91b7e5b..0000000
--- a/conda/meta.yaml
+++ /dev/null
@@ -1,51 +0,0 @@
-{% set data = load_setup_py_data() %}
-
-package:
-   name: intake
-   version:  {{ data['version'] }}
-
-source:
-   path: ..
-
-build:
-  number: {{ environ.get('GIT_DESCRIBE_NUMBER', 0) }}
-  entry_points:
-    {% for entry in data['entry_points']['console_scripts'] %}
-    - {{ entry }}
-    {% endfor %}
-  script: python setup.py install --single-version-externally-managed --record=record.txt
-  noarch: python
-
-requirements:
-  build:
-    - python
-    - jinja2
-  run:
-    - python >=3.6
-    - numpy
-    - msgpack-python
-    - msgpack-numpy
-    - requests
-    - tornado >=4.5.1
-    - jinja2
-    - pyyaml
-    - dask >=1.0
-    - appdirs
-    - fsspec
-
-test:
-  requires:
-    - pytest
-    - pytest-cov
-    - coverage
-    - coveralls
-  commands:
-    - intake list -h
-    - intake-server -h
-    - py.test --pyargs -v intake
-
-about:
-  home: {{ data['url'] }}
-  license: {{ data['license'] }}
-  license_file: LICENSE
-  summary: {{ data['description'] }}
diff --git a/docs/source/api_base.rst b/docs/source/api_base.rst
index c987bb4..49007bf 100644
--- a/docs/source/api_base.rst
+++ b/docs/source/api_base.rst
@@ -4,6 +4,7 @@ Base Classes
 This is a reference API class listing, useful mainly for developers.
 
 .. autosummary::
+   intake.source.base.DataSourceBase
    intake.source.base.DataSource
    intake.source.base.PatternMixin
    intake.source.base.AliasSource
diff --git a/docs/source/making-plugins.rst b/docs/source/making-plugins.rst
index 10a78a5..2ebfc1f 100644
--- a/docs/source/making-plugins.rst
+++ b/docs/source/making-plugins.rst
@@ -86,32 +86,42 @@ From the user perspective, all of the metadata should be loaded once the data so
 schema (after ``discover()``, ``read()``, ``to_dask()``, etc have been called).
 
 
-Subclassing ``intake.source.base.DataSource``
----------------------------------------------
+Subclassing ``intake.source.base.DataSourceBase``
+-------------------------------------------------
 
 Every Intake driver class should be a subclass of ``intake.source.base.DataSource``.
 The class should have the following attributes to identify itself:
 
-- ``name``: The short name of the driver.  This should be a valid python identifier.  You should not include the
+- ``name``: The short name of the driver.  This should be a valid python identifier.
+  You should not include the
   word ``intake`` in the driver name.
 
-- ``version``: A version string for the driver.  This may be reported to the user by tools based on Intake, but has
+- ``version``: A version string for the driver.  This may be reported to the user by tools
+  based on Intake, but has
   no semantic importance.
 
-- ``container``: The container type of data sources created by this object, e.g., ``dataframe``, ``ndarray``, or
+- ``container``: The container type of data sources created by this object, e.g.,
+  ``dataframe``, ``ndarray``, or
   ``python``, one of the keys of ``intake.container.container_map``.
-  For simplicity, a driver many only return one typed of container.  If a particular source of data could
-  be used in multiple ways (such as HDF5 files interpreted as dataframes or as ndarrays), two drivers must be created.
+  For simplicity, a driver many only return one typed of container.  If a particular
+  source of data could
+  be used in multiple ways (such as HDF5 files interpreted as dataframes or as ndarrays),
+  two drivers must be created.
   These two drivers can be part of the same Python package.
 
-- ``partition_access``: Do the data sources returned by this driver have multiple partitions?  This may help tools in
-  the future make more optimal decisions about how to present data.  If in doubt (or the answer depends on init
-  arguments), ``True`` will always result in correct behavior, even if the data source has only one partition.
+- ``partition_access``: Do the data sources returned by this driver have multiple
+  partitions?  This may help tools in
+  the future make more optimal decisions about how to present data.  If in doubt
+  (or the answer depends on init
+  arguments), ``True`` will always result in correct behavior, even if the data
+  source has only one partition.
 
-The ``__init()__`` method should always accept a keyword argument ``metadata``, a dictionary of metadata from the
+The ``__init()__`` method should always accept a keyword argument ``metadata``, a
+dictionary of metadata from the
 catalog to associate with the source.  This dictionary must be serializable as JSON.
 
-The base `DataSource` class has a small number of methods which should be overridden. Here is an example producing a
+The `DataSourceBase` class has a small number of methods which should be overridden.
+Here is an example producing a
 data-frame::
 
     class FooSource(intake.source.base.DataSource):
@@ -150,16 +160,18 @@ data-frame::
 Most of the work typically happens in the following methods:
 
 - ``__init__()``: Should be very lightweight and fast.  No files or network resources should be opened, and no
-  significant memory should be allocated yet.  Data sources are often serialized immediately.  The default implementation
+  significant memory should be allocated yet.  Data sources might be serialized immediately.  The default implementation
   of the pickle protocol in the base class will record all the arguments to ``__init__()`` and recreate the object with
   those arguments when unpickled, assuming the class has no side effects.
 
 - ``_get_schema()``: May open files and network resources and return as much of the schema as possible in small
-  amount of *approximately* constant  time.  The ``npartitions`` and ``extra_metadata`` attributes must be correct
+  amount of *approximately* constant  time. Typically, imports of packages needed by the source only happen here.
+  The ``npartitions`` and ``extra_metadata`` attributes must be correct
   when ``_get_schema`` returns.  Further keys such as ``dtype``, ``shape``, etc., should reflect the container type of
   the data-source, and can be ``None`` if not easily knowable, or include ``None`` for some elements. File-based
   sources should use fsspec to open a local or remote URL, and pass ``storage_options`` to it. This ensures
-  compatibility and extra features such as caching.
+  compatibility and extra features such as caching. If the backend can only deal with local files, you may
+  still want to use ``fsspec.open_local`` to allow for caching.
 
 - ``_get_partition(self, i)``: Should return all of the data from partition id ``i``, where ``i`` is typically an
   integer, but may be something more complex.
@@ -191,8 +203,28 @@ The full set of user methods of interest are as follows:
 - ``close(self)``: Close network or file handles and deallocate memory.  If other methods are called after ``close()``,
   the source is automatically reopened.
 
-It is also important to note that source attributes should be set after ``read()``, ``read_chunked()``,
-``read_partition()`` and ``to_dask()``, even if ``discover()`` was not called by the user.
+- ``to_*``: for some sources, it makes sense to provide alternative outputs aside from the base container
+  (dataframe, array, ...) and Dask variants.
+
+Note that all of these methods typically call ``_get_schema``, to make sure that the source has been
+initialised.
+
+Subclassing ``intake.source.base.DataSource``
+---------------------------------------------
+
+``DataSource`` provides the same functionality as ``DataSourceBase``, but has some additional mixin
+classes to provide some extras. A developer may choose to derive from ``DataSource`` to get all of
+these, or from ``DataSourceBase`` and make their own choice of mixins to support.
+
+- ``HoloviewsMixin``: provides plotting and GUI capabilities via the `holoviz`_ stack
+
+- ``PersistMixin``: allows for storing a local copy in a default format for the given
+  container type
+
+- ``CacheMixin``: allows for local storage of data files for a source. Deprecated,
+  you should use one of the caching mechanisms in ``fsspec``.
+
+.. _holoviz: https://holoviz.org/index.html
 
 .. _driver-discovery:
 
@@ -341,7 +373,7 @@ To use an ``OpenFile`` object, make it concrete by using a context:
 .. code-block:: python
 
     # at setup, to discover the number of files/partitions
-    set_of_open_files = dask.bytes.open_files(urlpath, mode='rb', **storage_options)
+    set_of_open_files = fsspec.open_files(urlpath, mode='rb', **storage_options)
 
     # when actually loading data; here we loop over all files, but maybe we just do one partition
     for an_open_file in set_of_open_files:
diff --git a/intake/__init__.py b/intake/__init__.py
index 27a3330..a661fce 100644
--- a/intake/__init__.py
+++ b/intake/__init__.py
@@ -17,7 +17,7 @@ del get_versions
 from .source import register_driver, registry
 
 imports = {
-    "DataSource": "intake.source.base:DataSoruce",
+    "DataSource": "intake.source.base:DataSource",
     'Schema': "intake.source.base:Schema",
     "cat": "intake.catalog:builtin",
     "load_combo_catalog": "intake.catalog.default:load_combo_catalog",
diff --git a/intake/catalog/base.py b/intake/catalog/base.py
index a096e8a..5b2c2b4 100644
--- a/intake/catalog/base.py
+++ b/intake/catalog/base.py
@@ -414,7 +414,7 @@ class Catalog(DataSource):
 
     def discover(self):
         return {"container": 'catalog', 'shape': None,
-                'dtype': None, 'datashape': None, 'metadata': self.metadata}
+                'dtype': None, 'metadata': self.metadata}
 
     def _close(self):
         # TODO: maybe close all entries?
diff --git a/intake/catalog/entry.py b/intake/catalog/entry.py
index 7b7a07f..e533fab 100644
--- a/intake/catalog/entry.py
+++ b/intake/catalog/entry.py
@@ -69,13 +69,13 @@ class CatalogEntry(DictSerialiseMixin):
             None, must be one of ['always', 'never', 'default']. Has no
             effect if the source has never been persisted, use source.persist()
         """
-
+        from ..source.base import PersistMixin
         if persist is not None and persist not in [
                 'always', 'never', 'default']:
             raise ValueError('Persist value (%s) not understood' % persist)
         persist = persist or self._pmode
         s = self.get(**kwargs)
-        if persist != 'never' and s.has_been_persisted:
+        if persist != 'never' and isinstance(s, PersistMixin) and s.has_been_persisted:
             from ..container.persist import store
 
             s2 = s.get_persisted()
diff --git a/intake/container/base.py b/intake/container/base.py
index bb55aa9..ceb98e0 100644
--- a/intake/container/base.py
+++ b/intake/container/base.py
@@ -55,7 +55,6 @@ class RemoteSource(DataSource):
             self._parse_open_response(response)
 
     def _parse_open_response(self, response):
-        self.datashape = response['datashape']
         dtype_descr = response['dtype']
         if isinstance(dtype_descr, list):
             # Reformat because NumPy needs list of tuples
diff --git a/intake/container/dataframe.py b/intake/container/dataframe.py
index 93f5a4a..019435b 100644
--- a/intake/container/dataframe.py
+++ b/intake/container/dataframe.py
@@ -27,8 +27,7 @@ class RemoteDataFrame(RemoteSource):
         self._schema = Schema(npartitions=self.npartitions,
                               extra_metadata=self.metadata,
                               dtype=self.dtype,
-                              shape=self.shape,
-                              datashape=None)
+                              shape=self.shape)
         self.dataframe = None
 
     def _load_metadata(self):
@@ -167,8 +166,7 @@ class GenericDataFrame(DataSource):
             self._schema = Schema(npartitions=self.npartitions,
                                   extra_metadata=self.metadata,
                                   dtype=self.dtype,
-                                  shape=self.shape,
-                                  datashape=None)
+                                  shape=self.shape)
         return self._schema
 
     def _get_partition(self, i):
diff --git a/intake/container/ndarray.py b/intake/container/ndarray.py
index 4f9cc2b..4d9e8eb 100644
--- a/intake/container/ndarray.py
+++ b/intake/container/ndarray.py
@@ -28,8 +28,7 @@ class RemoteArray(RemoteSource):
                               extra_metadata=self.metadata,
                               dtype=self.dtype,
                               shape=self.shape,
-                              chunks=self.chunks,
-                              datashape=None)
+                              chunks=self.chunks)
 
     def _load_metadata(self):
         import dask.array as da
diff --git a/intake/source/base.py b/intake/source/base.py
index 335d90a..98de12a 100644
--- a/intake/source/base.py
+++ b/intake/source/base.py
@@ -11,7 +11,6 @@ from yaml import dump
 
 from .cache import make_caches
 from ..utils import make_path_posix, DictSerialiseMixin, pretty_describe
-import warnings
 
 
 class Schema(dict):
@@ -26,10 +25,10 @@ class Schema(dict):
         """
         Parameters
         ----------
-        kwargs: typically include datashape, dtype, shape
+        kwargs: typically include dtype, shape
         """
         super(Schema, self).__init__(**kwargs)
-        for field in ['datashape', 'dtype', 'shape']:
+        for field in ['dtype', 'shape']:
             # maybe a default-dict
             if field not in self:
                 self[field] = None
@@ -53,7 +52,138 @@ class NoEntry(AttributeError):
     pass
 
 
-class DataSource(DictSerialiseMixin):
+class CacheMixin:
+    """Allows "old style" caching for Data Source"""
+    _cache = None
+
+    @property
+    def cache_dirs(self):
+        return [c._cache_dir for c in self.cache]
+
+    def set_cache_dir(self, cache_dir):
+        for c in self.cache:
+            c._cache_dir = make_path_posix(cache_dir)
+
+    @property
+    def cache(self):
+        if self._cache is None:
+            self._cache = make_caches(self.name, self.metadata.get('cache'),
+                                      catdir=self.metadata.get('catalog_dir',
+                                                             None),
+                                      storage_options=self.storage_options)
+        return self._cache
+
+    @cache.setter
+    def cache(self, csh):
+        self._cache = csh
+
+    def _get_cache(self, urlpath):
+        if len(self.cache) == 0:
+            return [urlpath]
+        return [c.load(urlpath) for c in self.cache]
+
+
+class HoloviewsMixin:
+    """Adds plotting and GUI to DataSource"""
+
+    @property
+    def plots(self):
+        """List custom associated quick-plots """
+        return list(self.metadata.get('plots', {}))
+
+    @property
+    def gui(self):
+        """Source GUI, with parameter selection and plotting"""
+        return self.entry.gui
+
+    @property
+    def plot(self):
+        """
+        Returns a hvPlot object to provide a high-level plotting API.
+
+        To display in a notebook, be sure to run ``intake.output_notebook()``
+        first.
+        """
+        try:
+            from hvplot import hvPlot
+        except ImportError:
+            raise ImportError("The intake plotting API requires hvplot."
+                              "hvplot may be installed with:\n\n"
+                              "`conda install -c pyviz hvplot` or "
+                              "`pip install hvplot`.")
+        metadata = self.metadata.get('plot', {})
+        fields = self.metadata.get('fields', {})
+        for attrs in fields.values():
+            if 'range' in attrs:
+                attrs['range'] = tuple(attrs['range'])
+        metadata['fields'] = fields
+        plots = self.metadata.get('plots', {})
+        return hvPlot(self, custom_plots=plots, **metadata)
+
+    @property
+    def hvplot(self):
+        """
+        Returns a hvPlot object to provide a high-level plotting API.
+        """
+        return self.plot
+
+
+class PersistMixin:
+    """Adds interaction with PersistStore to DataSource"""
+
+    def get_persisted(self):
+        from ..container.persist import store
+        from dask.base import tokenize
+        return store[tokenize(self)]()
+
+    @staticmethod
+    def _persist(source, path, **kwargs):
+        """To be implemented by 'container' sources for locally persisting"""
+        raise NotImplementedError
+
+    @property
+    def has_been_persisted(self):
+        from ..container.persist import store
+        from dask.base import tokenize
+        return tokenize(self) in store
+
+    @property
+    def is_persisted(self):
+        from ..container.persist import store
+        return self.metadata.get('original_tok', None) in store
+
+    def persist(self, ttl=None, **kwargs):
+        """Save data from this source to local persistent storage
+
+        Parameters
+        ----------
+        ttl: numeric, optional
+            Time to live in seconds. If provided, the original source will
+            be accessed and a new persisted version written transparently
+            when more than ``ttl`` seconds have passed since the old persisted
+            version was written.
+        kargs: passed to the _persist method on the base container.
+        """
+        from ..container.persist import PersistStore
+        from dask.base import tokenize
+        if 'original_tok' in self.metadata:
+            raise ValueError('Cannot persist a source taken from the persist '
+                             'store')
+        if ttl is not None and not isinstance(ttl, (int, float)):
+            raise ValueError('Cannot persist using a time to live that is '
+                             f'non-numeric. User-provided ttl was {ttl}')
+        store = PersistStore()
+        out = self._export(store.getdir(self), **kwargs)
+        out.metadata.update({
+            'ttl': ttl,
+            'cat': {} if self.cat is None else self.cat.__getstate__()
+        })
+        out.name = self.name
+        store.add(tokenize(self), out)
+        return out
+
+
+class DataSourceBase(DictSerialiseMixin):
     """An object which can produce data
 
     This is the base class for all Intake plugins, including catalogs and
@@ -62,20 +192,19 @@ class DataSource(DictSerialiseMixin):
 
     This class is not useful in itself, most methods raise NotImplemented.
     """
+
     name = None
     version = None
     container = None
     partition_access = False
-    datashape = None
     description = None
-
-    @property
-    def cache_dirs(self):
-        return [c._cache_dir for c in self.cache]
-
-    def set_cache_dir(self, cache_dir):
-        for c in self.cache:
-            c._cache_dir = make_path_posix(cache_dir)
+    dtype = None
+    shape = None
+    npartitions = 0
+    _schema = None
+    on_server = False
+    cat = None  # the cat from which this source was made
+    _entry = None
 
     def __init__(self, storage_options=None, metadata=None):
         # default data
@@ -83,82 +212,55 @@ class DataSource(DictSerialiseMixin):
         if isinstance(self.metadata, dict):
             storage_options = self._captured_init_kwargs.get('storage_options',
                                                              {})
-            self.cache = make_caches(self.name, self.metadata.get('cache'),
-                                     catdir=self.metadata.get('catalog_dir',
-                                                              None),
-                                     storage_options=storage_options)
-        self.datashape = None
-        self.dtype = None
-        self.shape = None
-        self.npartitions = 0
-        self._schema = None
-        self.catalog_object = None
-        self.on_server = False
-        self.cat = None  # the cat from which this source was made
-        self._entry = None
-
-    def _get_cache(self, urlpath):
-        if len(self.cache) == 0:
-            return [urlpath]
-        return [c.load(urlpath) for c in self.cache]
+        self.storage_options = storage_options
 
     def _get_schema(self):
         """Subclasses should return an instance of base.Schema"""
-        raise Exception('Subclass should implement _get_schema()')
+        raise NotImplementedError
 
     def _get_partition(self, i):
         """Subclasses should return a container object for this partition
 
         This function will never be called with an out-of-range value for i.
         """
-        raise Exception('Subclass should implement _get_partition()')
+        raise NotImplementedError
 
     def _close(self):
         """Subclasses should close all open resources"""
-        raise Exception('Subclass should implement _close()')
-
-    # These methods are implemented from the above two methods and do not need
-    # to be overridden unless custom behavior is required
+        raise NotImplementedError
 
     def _load_metadata(self):
         """load metadata only if needed"""
         if self._schema is None:
             self._schema = self._get_schema()
-            self.datashape = self._schema.datashape
             self.dtype = self._schema.dtype
             self.shape = self._schema.shape
             self.npartitions = self._schema.npartitions
             self.metadata.update(self._schema.extra_metadata)
 
-    def _yaml(self, with_plugin=False):
+    def _yaml(self):
         import inspect
         kwargs = self._captured_init_kwargs.copy()
         meta = kwargs.pop('metadata', self.metadata) or {}
         kwargs.update(dict(zip(inspect.signature(self.__init__).parameters,
                            self._captured_init_args)))
-        data = {'sources':
+        data = {
+            'sources':
                 {self.name: {
-            'driver': self.classname,
-            'description': self.description or "",
-            'metadata': meta,
-            'args': kwargs
-        }}}
+                   'driver': self.classname,
+                   'description': self.description or "",
+                   'metadata': meta,
+                   'args': kwargs
+                }}}
         return data
 
-    def yaml(self, with_plugin=False):
+    def yaml(self):
         """Return YAML representation of this data-source
 
         The output may be roughly appropriate for inclusion in a YAML
         catalog. This is a best-effort implementation
-
-        Parameters
-        ----------
-        with_plugin: bool
-            If True, create a "plugins" section, for cases where this source
-            is created with a plugin not expected to be in the global Intake
-            registry.
         """
-        data = self._yaml(with_plugin=with_plugin)
+        data = self._yaml()
         return dump(data, default_flow_style=False)
 
     def _ipython_display_(self):
@@ -177,16 +279,24 @@ class DataSource(DictSerialiseMixin):
         return self.yaml()
 
     @property
-    def plots(self):
-        """List custom associated quick-plots """
-        return list(self.metadata.get('plots', {}))
+    def is_persisted(self):
+        """The base class does not interact with persistence"""
+        return False
+
+    @property
+    def has_been_persisted(self):
+        """The base class does not interact with persistence"""
+        return False
+
+    def _get_cache(self, urlpath):
+        """The base class does not interact with caches"""
+        return urlpath
 
     def discover(self):
         """Open resource and populate the source attributes."""
         self._load_metadata()
 
-        return dict(datashape=self.datashape,
-                    dtype=self.dtype,
+        return dict(dtype=self.dtype,
                     shape=self.shape,
                     npartitions=self.npartitions,
                     metadata=self.metadata)
@@ -234,7 +344,6 @@ class DataSource(DictSerialiseMixin):
 
     @property
     def entry(self):
-        warnings.warn("direct access to the entry is deprecated", stacklevel=2)
         if self._entry is None:
             raise NoEntry("Source was not made from a catalog entry")
         return self._entry
@@ -262,18 +371,8 @@ class DataSource(DictSerialiseMixin):
     __call__ = get = configure_new  # compatibility aliases
 
     def describe(self):
-        if self._entry is None:
-            raise NoEntry("Source was not made from a catalog entry")
-
-        return self._entry.describe()
-
-    @property
-    def gui(self):
-        """Source GUI, with parameter selection and plotting"""
-        if self._entry is None:
-            raise NoEntry("Source was not made from a catalog entry")
-
-        return self._entry.gui
+        """Description from the entry spec"""
+        return self.entry.describe()
 
     def close(self):
         """Close open resources corresponding to this data source."""
@@ -287,67 +386,6 @@ class DataSource(DictSerialiseMixin):
     def __exit__(self, exc_type, exc_value, traceback):
         self.close()
 
-    @property
-    def plot(self):
-        """
-        Returns a hvPlot object to provide a high-level plotting API.
-
-        To display in a notebook, be sure to run ``intake.output_notebook()``
-        first.
-        """
-        try:
-            from hvplot import hvPlot
-        except ImportError:
-            raise ImportError("The intake plotting API requires hvplot."
-                              "hvplot may be installed with:\n\n"
-                              "`conda install -c pyviz hvplot` or "
-                              "`pip install hvplot`.")
-        metadata = self.metadata.get('plot', {})
-        fields = self.metadata.get('fields', {})
-        for attrs in fields.values():
-            if 'range' in attrs:
-                attrs['range'] = tuple(attrs['range'])
-        metadata['fields'] = fields
-        plots = self.metadata.get('plots', {})
-        return hvPlot(self, custom_plots=plots, **metadata)
-
-    @property
-    def hvplot(self):
-        """
-        Returns a hvPlot object to provide a high-level plotting API.
-        """
-        return self.plot
-
-    def persist(self, ttl=None, **kwargs):
-        """Save data from this source to local persistent storage
-
-        Parameters
-        ----------
-        ttl: numeric, optional
-            Time to live in seconds. If provided, the original source will
-            be accessed and a new persisted version written transparently
-            when more than ``ttl`` seconds have passed since the old persisted
-            version was written.
-        kargs: passed to the _persist method on the base container.
-        """
-        from ..container.persist import PersistStore
-        from dask.base import tokenize
-        if 'original_tok' in self.metadata:
-            raise ValueError('Cannot persist a source taken from the persist '
-                             'store')
-        if ttl is not None and not isinstance(ttl, (int, float)):
-            raise ValueError('Cannot persist using a time to live that is '
-                             f'non-numeric. User-provided ttl was {ttl}')
-        store = PersistStore()
-        out = self._export(store.getdir(self), **kwargs)
-        out.metadata.update({
-            'ttl': ttl,
-            'cat': {} if self.cat is None else self.cat.__getstate__()
-        })
-        out.name = self.name
-        store.add(tokenize(self), out)
-        return out
-
     def export(self, path, **kwargs):
         """Save this data for sharing with other people
 
@@ -378,26 +416,14 @@ class DataSource(DictSerialiseMixin):
         out.name = self.name
         return out
 
-    def get_persisted(self):
-        from ..container.persist import store
-        from dask.base import tokenize
-        return store[tokenize(self)]()
 
-    @staticmethod
-    def _persist(source, path, **kwargs):
-        """To be implemented by 'container' sources for locally persisting"""
-        raise NotImplementedError
-
-    @property
-    def has_been_persisted(self):
-        from ..container.persist import store
-        from dask.base import tokenize
-        return tokenize(self) in store
+class DataSource(CacheMixin, HoloviewsMixin, PersistMixin, DataSourceBase):
+    """A Data Source will all optional functionality
 
-    @property
-    def is_persisted(self):
-        from ..container.persist import store
-        return self.metadata.get('original_tok', None) in store
+    When subclassed, child classes will have the base data source functionality,
+    plus caching, plotting and persistence abilities.
+    """
+    pass
 
 
 class PatternMixin(object):
@@ -488,16 +514,15 @@ class AliasSource(DataSource):
         self.source = None
 
     def _get_source(self):
-        if self.catalog_object is None:
+        if self.cat is None:
             raise ValueError('AliasSource cannot be used outside a catalog')
         if self.source is None:
-            self.source = self.catalog_object[self.mapping[self.target]](
+            self.source = self.cat[self.mapping[self.target]](
                 metadata=self.metadata, **self.kwargs)
             self.metadata = self.source.metadata.copy()
             self.container = self.source.container
             self.partition_access = self.source.partition_access
             self.description = self.source.description
-            self.datashape = self.source.datashape
 
     def discover(self):
         self._get_source()
diff --git a/intake/source/cache.py b/intake/source/cache.py
index 571998e..9b0c029 100644
--- a/intake/source/cache.py
+++ b/intake/source/cache.py
@@ -17,7 +17,6 @@ import re
 import shutil
 import warnings
 
-from fsspec.utils import infer_storage_options
 from intake.config import conf
 from intake.utils import make_path_posix
 
@@ -26,6 +25,7 @@ logger = logging.getLogger('intake')
 
 def sanitize_path(path):
     """Utility for cleaning up paths."""
+    from fsspec.utils import infer_storage_options
 
     storage_option = infer_storage_options(path)
 
diff --git a/intake/source/csv.py b/intake/source/csv.py
index d497a09..77eaad3 100644
--- a/intake/source/csv.py
+++ b/intake/source/csv.py
@@ -128,8 +128,7 @@ class CSVSource(base.DataSource, base.PatternMixin):
 
         dtypes = self._dataframe._meta.dtypes.to_dict()
         dtypes = {n: str(t) for (n, t) in dtypes.items()}
-        return base.Schema(datashape=None,
-                           dtype=dtypes,
+        return base.Schema(dtype=dtypes,
                            shape=(None, len(dtypes)),
                            npartitions=self._dataframe.npartitions,
                            extra_metadata={})
diff --git a/intake/source/textfiles.py b/intake/source/textfiles.py
index de55319..1c20c00 100644
--- a/intake/source/textfiles.py
+++ b/intake/source/textfiles.py
@@ -81,8 +81,7 @@ class TextFilesSource(base.DataSource):
                 compression=self.compression,
                 **self._storage_options)
             self.npartitions = len(self._files)
-        return base.Schema(datashape=None,
-                           dtype=None,
+        return base.Schema(dtype=None,
                            shape=(None, ),
                            npartitions=self.npartitions,
                            extra_metadata=self.metadata)
diff --git a/intake/utils.py b/intake/utils.py
index ef0dd25..c3a714e 100644
--- a/intake/utils.py
+++ b/intake/utils.py
@@ -6,6 +6,7 @@
 #-----------------------------------------------------------------------------
 
 import collections
+from collections import OrderedDict
 import collections.abc
 import datetime
 from contextlib import contextmanager
@@ -47,13 +48,14 @@ def tuple_constructor(loader, node, deep=False):
     return tuple(loader.construct_object(node, deep=deep)
                  for node in node.value)
 
-from collections import OrderedDict
 
 def represent_dictionary_order(self, dict_data):
     return self.represent_mapping('tag:yaml.org,2002:map', dict_data.items())
 
+
 yaml.add_representer(OrderedDict, represent_dictionary_order)
 
+
 @contextmanager
 def no_duplicate_yaml():
     yaml.SafeLoader.add_constructor(
@@ -89,14 +91,6 @@ class DictSerialiseMixin(object):
 
     __tok_cache = None
 
-    @property
-    def _tok(self):
-        import warnings
-        from dask.base import tokenize
-        warnings.warn("the _tok attribute is deprecated, please use "
-                      "`dask.base.tokenize(obj)` instead")
-        return tokenize(self)
-
     def __new__(cls, *args, **kwargs):
         """Capture creation args when instantiating"""
         o = object.__new__(cls)
@@ -170,6 +164,7 @@ def pretty_describe(object, nestedness=0, indent=2):
         return f'{sep}{out}'
     return out
 
+
 def decode_datetime(obj):
     import numpy
     if not isinstance(obj, numpy.ndarray) and "__datetime__" in obj:
@@ -185,6 +180,7 @@ def decode_datetime(obj):
             )
     return obj
 
+
 def encode_datetime(obj):
     if isinstance(obj, datetime.datetime):
         return {"__datetime__": True, "as_str": obj.strftime("%Y%m%dT%H:%M:%S.%f%z")}
@@ -252,16 +248,16 @@ class DriverRegistryView(RegistryView):
     # This attributes are used by the base class
     # to fill in warning and error messages.
     _registry_name = "intake.registry"
-    _register_func_name  = "intake.register_driver"
-    _unregister_func_name  = "intake.unregister_driver"
+    _register_func_name = "intake.register_driver"
+    _unregister_func_name = "intake.unregister_driver"
 
 
 class ContainerRegistryView(RegistryView):
     # This attributes are used by the base class
     # to fill in warning and error messages.
     _registry_name = "intake.container_map"
-    _register_func_name  = "intake.register_container"
-    _unregister_func_name  = "intake.unregister_container"
+    _register_func_name = "intake.register_container"
+    _unregister_func_name = "intake.unregister_container"
 
 
 class ModuleImporter:
diff --git a/requirements.txt b/requirements.txt
index f4c6ccb..d013830 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,11 +1,5 @@
-# This file is also processed by conda-build, so keep spaces as "pkg >=1.2"
 appdirs
+dask
 entrypoints
-dask[bag] >=1.0
-jinja2
-numpy
 pyyaml
-requests
-msgpack
 fsspec >=0.7.4
-aiohttp


Test patch:
diff --git a/intake/catalog/tests/test_remote_integration.py b/intake/catalog/tests/test_remote_integration.py
index cfc00e7..5717324 100644
--- a/intake/catalog/tests/test_remote_integration.py
+++ b/intake/catalog/tests/test_remote_integration.py
@@ -122,7 +122,6 @@ def test_read(intake_server):
     meta = expected_df[:0]
 
     info = d.discover()
-    assert info['datashape'] is None
     assert info['dtype'] == {k: str(v) for k, v
                              in meta.dtypes.to_dict().items()}
     assert info['npartitions'] == 2
@@ -148,7 +147,6 @@ def test_read_direct(intake_server):
 
     info = d.discover()
 
-    assert info['datashape'] is None
     assert info['dtype'] == {k: str(v) for k, v
                              in meta.dtypes.to_dict().items()}
     assert info['npartitions'] == 1
diff --git a/intake/cli/client/tests/test_local_integration.py b/intake/cli/client/tests/test_local_integration.py
index 24d41ed..0f95a9d 100644
--- a/intake/cli/client/tests/test_local_integration.py
+++ b/intake/cli/client/tests/test_local_integration.py
@@ -86,7 +86,6 @@ def test_discover():
                                universal_newlines=True)
     out, _ = process.communicate()
 
-    assert "'datashape':" in out
     assert "'dtype':" in out
     assert "'metadata':" in out
     assert "'npartitions':" in out
diff --git a/intake/source/tests/test_base.py b/intake/source/tests/test_base.py
index acc6178..e214416 100644
--- a/intake/source/tests/test_base.py
+++ b/intake/source/tests/test_base.py
@@ -23,9 +23,8 @@ def test_datasource_base_method_exceptions():
                                 ('_get_partition', [1]),
                                 ('_close', [])]:
         method = getattr(d, method_name)
-        with pytest.raises(Exception) as except_info:
+        with pytest.raises(NotImplementedError):
             method(*args)
-        assert method_name in str(except_info.value)
 
 
 def test_name():
@@ -38,10 +37,9 @@ def test_datasource_base_context_manager():
     # Base data source should raise a "need to implement" exception when it
     # enters the context manager (which loads the schema)
 
-    with pytest.raises(Exception) as except_info:
+    with pytest.raises(NotImplementedError):
         with base.DataSource():
             pass
-    assert '_get_schema' in str(except_info.value)
 
 
 class MockDataSourceDataFrame(base.DataSource):
@@ -67,8 +65,7 @@ class MockDataSourceDataFrame(base.DataSource):
     def _get_schema(self):
         self.call_count['_get_schema'] += 1
 
-        return base.Schema(datashape='datashape',
-                           dtype=np.dtype([('x', np.int64), ('y', np.int64)]),
+        return base.Schema(dtype=np.dtype([('x', np.int64), ('y', np.int64)]),
                            shape=(6,),
                            npartitions=2,
                            extra_metadata=dict(c=3, d=4))
@@ -109,7 +106,6 @@ def test_datasource_discover(source_dataframe):
 
     row_dtype = np.dtype([('x', np.int64), ('y', np.int64)])
     assert r == {
-        'datashape': 'datashape',
         'dtype': row_dtype,
         'shape': (6,),
         'npartitions': 2,
@@ -117,7 +113,6 @@ def test_datasource_discover(source_dataframe):
     }
 
     # check attributes have been set
-    assert source_dataframe.datashape == 'datashape'
     assert source_dataframe.dtype == row_dtype
     assert source_dataframe.shape == (6,)
     assert source_dataframe.npartitions == 2
@@ -232,8 +227,7 @@ class MockDataSourcePython(base.DataSource):
     def _get_schema(self):
         self.call_count['_get_schema'] += 1
 
-        return base.Schema(datashape=None,
-                           dtype=None,
+        return base.Schema(dtype=None,
                            shape=(4,),
                            npartitions=2,
                            extra_metadata=dict(c=3, d=4))
@@ -274,7 +268,6 @@ def test_datasource_python_discover(source_python):
     assert source_python.container == 'python'
 
     assert r == {
-        'datashape': None,
         'dtype': None,
         'shape': (4,),
         'npartitions': 2,
@@ -282,7 +275,6 @@ def test_datasource_python_discover(source_python):
     }
 
     # check attributes have been set
-    assert source_python.datashape is None
     assert source_python.dtype is None
     assert source_python.shape == (4,)
     assert source_python.npartitions == 2
diff --git a/intake/source/tests/util.py b/intake/source/tests/util.py
index bbfe1e7..e783f15 100644
--- a/intake/source/tests/util.py
+++ b/intake/source/tests/util.py
@@ -13,7 +13,7 @@ def verify_plugin_interface(plugin):
 
 
 def verify_datasource_interface(source):
-    for attr in ['container', 'description', 'datashape', 'dtype', 'shape',
+    for attr in ['container', 'description', 'dtype', 'shape',
                  'npartitions', 'metadata']:
         assert hasattr(source, attr)
 
diff --git a/test_requirements.txt b/test_requirements.txt
index c16f90f..f636241 100644
--- a/test_requirements.txt
+++ b/test_requirements.txt
@@ -4,3 +4,6 @@ notebook
 panel==0.8.0
 hvplot==0.5.2
 bokeh<2.0.0
+fsspec
+aiohttp
+requests