Repository: dask/dask
Pull request: Add persist function
This is an initial step to resolve #1908 

This copies over some code from dask/distributed to implement a persist function that uses a get function to compute results and re-insert those results back into a Dask collection with the same keys.

Short term I'm likely to take this only as far as I need to accomplish current work.  If anyone else wants to jump in with longer-term thinking and commit directly to this branch they are most welcome. cc @jcrist 

Problem statement:
Add persist function
The distributed scheduler makes heavy use of `persist`, which creates a new dask collection with a collection that points to more evaluated data.  

It might be convenient to add this verb to dask generally.

```python
df = dd.read_csv(...)  # lazy loading
df = dask.persist(df)  # df is dask.dataframe that points to pandas dataframes
```

Persist would be a blocking call (to support general dask schedulers) that in the distributed case would be the same as 

```python
def persist(...)
    collection = global_client.persist(collection)
    wait(collection)
    return collection
```

In the single machine case it would probably be more like the following:

```python
def persist(...):
    keys = collection._keys()
    evaluated = get(collection.dask, keys)
    dsk = zip(keys, evaluated)
    return redict_collection(collection, dsk)

def redict_collection(c, dsk):  # this was taken from distributed/client.py
    if isinstance(c, Delayed):
        return Delayed(c.key, [dsk])
    else:
        cc = copy.copy(c)
        cc.dask = dsk
        return cc
```

Having a general `persist` function would make it easier to build algorithms that were consistent between single and distributed schedulers, such as occurs in dask-glm (cc @moody-marlin).  It would also help users who want to use dask for parallelism needs rather than out-of-core computation (cc @jbednar)

@jcrist @shoyer thoughts?

Hints from issue discussion:
I like the idea. It may be nice to also support a general mutable-mapping to persist to, to allow for spilling to disk. I could see a few options for how to do this:

- Set persist mapping globally using `set_options`
- Pass a mapping to the persist call `persist(collection, cache=cache)`

The graph returned from `persist` might then look like `{new_key: (getitem, cache, key), ...}` instead of just `{new_key: result_of_old_key}`.

---

Alternatively, persist could be a method on a context instead of a function. Something like:
    
```python
# Needs a better name than `Context`
context = Context(cache=my_spill_to_disk_dictionary_thing)
result = context.persist(collection)
# The compute method would then pull results from the local
# cache before computing
context.compute(other_collection_maybe_based_off_this_one)
```

One perk of this instead of a global/parameter method is that it mirrors the distributed client context, and would allow for a mimicking of the incremental computation api distributed allows. The global compute/persist could then just be

```python
global_context = Context(cache={})
compute = global_context.compute
persist = global_context.persist
```

This isn't as simple as a global thing though, and I'm unsure that it's needed. Just a thought.
I guess this would help to organize the current set of globals that we keep around in `set_options`.  

@jcrist do you have any desire/time to experiment with implementing this?
Yes, I think this is a good idea. With xarray, I found myself writing things like `.compute().chunk(...)` with some frequency, both for parallelism and to solve scheduler fail cases like https://github.com/dask/dask/issues/874. Persisting to a dictionary with automatic spilling to disk could make this very convenient.
Yeah, I'd be interested in working on this. I'm trying to finish up the categorical work today, then should have some time to work on this.

Commit message: flake8

Patch to review:
diff --git a/dask/__init__.py b/dask/__init__.py
index f5f266a6883..5da166f83dd 100644
--- a/dask/__init__.py
+++ b/dask/__init__.py
@@ -8,7 +8,7 @@
 except ImportError:
     pass
 try:
-    from .base import visualize, compute
+    from .base import visualize, compute, persist
 except ImportError:
     pass
 
diff --git a/dask/base.py b/dask/base.py
index 2f24db9006d..30e6df08b07 100644
--- a/dask/base.py
+++ b/dask/base.py
@@ -1,9 +1,9 @@
 from __future__ import absolute_import, division, print_function
 
 from collections import OrderedDict
+import copy
 from functools import partial
 from hashlib import md5
-from operator import attrgetter
 import pickle
 import os
 import uuid
@@ -13,6 +13,7 @@
 
 from .compatibility import bind_method, unicode
 from .context import _globals
+from .core import flatten
 from .utils import Dispatch
 
 __all__ = ("Base", "compute", "normalize_token", "tokenize", "visualize")
@@ -61,8 +62,40 @@ def visualize(self, filename='mydask', format=None, optimize_graph=False,
         return visualize(self, filename=filename, format=format,
                          optimize_graph=optimize_graph, **kwargs)
 
+    def persist(self, **kwargs):
+        """ Persist this dask collection into memory
+
+        This turns a lazy Dask collection into a Dask collection with the same
+        metadata, but now with its results actively computing.  For example a
+        lazy dask.array built up from many lazy calls will now be a dask.array
+        of the same shape, dtype, etc., but now with all of those lazy
+        operations either in memory (for single-machine computing) or actively
+        computing asynchronously (for distributed computing).
+
+        If using Dask on a single machine then you should ensure that the
+        dataset fits entirely within memory.
+
+        Parameters
+        ----------
+        get : callable, optional
+            A scheduler ``get`` function to use. If not provided, the default
+            is to check the global settings first, and then fall back to
+            the collection defaults.
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before computation.
+            Otherwise the graph is run as is. This can be useful for debugging.
+        kwargs
+            Extra keywords to forward to the scheduler ``get`` function.
+        """
+        return persist(self, **kwargs)[0]
+
     def compute(self, **kwargs):
-        """Compute several dask collections at once.
+        """ Compute this dask collection
+
+        This turns a lazy Dask collection into its in-memory equivalent.
+        For example a Dask.array turns into a NumPy array and a Dask.dataframe
+        turns into a Pandas dataframe.  The entire dataset must fit into memory
+        before calling this operation.
 
         Parameters
         ----------
@@ -119,24 +152,6 @@ def _get_binary_operator(cls, op, inv=False):
         raise NotImplementedError
 
 
-def _extract_graph_and_keys(vals):
-    """Given a list of dask vals, return a single graph and a list of keys such
-    that ``get(dsk, keys)`` is equivalent to ``[v.compute() v in vals]``."""
-    dsk = {}
-    keys = []
-    for v in vals:
-        # Optimization to avoid merging dictionaries in Delayed values. Reduces
-        # memory usage for large graphs.
-        if hasattr(v, '_dasks'):
-            for d in v._dasks:
-                dsk.update(d)
-        else:
-            dsk.update(v.dask)
-        keys.append(v._keys())
-
-    return dsk, keys
-
-
 def compute(*args, **kwargs):
     """Compute several dask collections at once.
 
@@ -165,13 +180,12 @@ def compute(*args, **kwargs):
     >>> compute(a, b)
     (45, 4.5)
     """
+    optimize_graph = kwargs.pop('optimize_graph', True)
     variables = [a for a in args if isinstance(a, Base)]
     if not variables:
         return args
 
     get = kwargs.pop('get', None) or _globals['get']
-    optimizations = (kwargs.pop('optimizations', None) or
-                     _globals.get('optimizations', []))
 
     if not get:
         get = variables[0]._default_get
@@ -181,18 +195,8 @@ def compute(*args, **kwargs):
                              "scheduler `get` function using either "
                              "the `get` kwarg or globally with `set_options`.")
 
-    if kwargs.get('optimize_graph', True):
-        groups = groupby(attrgetter('_optimize'), variables)
-        groups = {opt: _extract_graph_and_keys(val)
-                  for opt, val in groups.items()}
-        for opt in optimizations:
-            groups = {k: [opt(dsk, keys), keys]
-                      for k, (dsk, keys) in groups.items()}
-        dsk = merge([opt(dsk, keys, **kwargs)
-                    for opt, (dsk, keys) in groups.items()])
-        keys = [var._keys() for var in variables]
-    else:
-        dsk, keys = _extract_graph_and_keys(variables)
+    dsk = collections_to_dsk(variables, optimize_graph, **kwargs)
+    keys = [var._keys() for var in variables]
     results = get(dsk, keys, **kwargs)
 
     results_iter = iter(results)
@@ -383,3 +387,151 @@ def tokenize(*args, **kwargs):
     if kwargs:
         args = args + (kwargs,)
     return md5(str(tuple(map(normalize_token, args))).encode()).hexdigest()
+
+
+def collections_to_dsk(collections, optimize_graph=True, **kwargs):
+    """
+    Convert many collections into a single dask graph, after optimization
+    """
+    optimizations = (kwargs.pop('optimizations', None) or
+                     _globals.get('optimizations', []))
+    if optimize_graph:
+        groups = groupby(lambda x: x._optimize, collections)
+        groups = {opt: _extract_graph_and_keys(val)
+                  for opt, val in groups.items()}
+        for opt in optimizations:
+            groups = {k: [opt(dsk, keys), keys]
+                      for k, (dsk, keys) in groups.items()}
+        dsk = merge([opt(dsk, keys, **kwargs)
+                     for opt, (dsk, keys) in groups.items()])
+    else:
+        dsk = merge(c.dask for c in collections)
+
+    return dsk
+
+
+def _extract_graph_and_keys(vals):
+    """Given a list of dask vals, return a single graph and a list of keys such
+    that ``get(dsk, keys)`` is equivalent to ``[v.compute() v in vals]``."""
+    dsk = {}
+    keys = []
+    for v in vals:
+        # Optimization to avoid merging dictionaries in Delayed values. Reduces
+        # memory usage for large graphs.
+        if hasattr(v, '_dasks'):
+            for d in v._dasks:
+                dsk.update(d)
+        else:
+            dsk.update(v.dask)
+        keys.append(v._keys())
+
+    return dsk, keys
+
+
+def redict_collection(c, dsk):
+    from dask.delayed import Delayed
+    if isinstance(c, Delayed):
+        return Delayed(c.key, [dsk])
+    else:
+        cc = copy.copy(c)
+        cc.dask = dsk
+        return cc
+
+
+def persist(*args, **kwargs):
+    """ Persist multiple Dask collections into memory
+
+    This turns lazy Dask collections into Dask collections with the same
+    metadata, but now with their results actively computing.
+
+    For example a lazy dask.array built up from many lazy calls will now be a
+    dask.array of the same shape, dtype, etc., but now with all of those lazy
+    operations either in memory (for single-machine computing) or actively
+    computing asynchronously (for distributed computing).
+
+    If using Dask on a single machine then you should ensure that the
+    dataset fits entirely within memory.
+
+    This function operates differently if a ``dask.distributed.Client`` exists
+    and is connected to a distributed scheduler.  In this case this function
+    will return as soon as the task graph has been submitted to the cluster,
+    but before the computations have completed.  Computations will continue
+    asynchronously in the background.  When using this function with the single
+    machine scheduler it blocks until the computations have finished.
+
+    Examples
+    --------
+
+    >>> df = dd.read_csv('/path/to/*.csv')  # doctest: +SKIP
+    >>> df = df[df.name == 'Alice']  # doctest: +SKIP
+    >>> df['in-debt'] = df.balance < 0  # doctest: +SKIP
+    >>> df = df.persist()  # triggers computation  # doctest: +SKIP
+
+    >>> df.value().min()  # future computations are now fast  # doctest: +SKIP
+    -10
+    >>> df.value().max()  # doctest: +SKIP
+    100
+
+    >>> from dask import persist  # use persist function on multiple collections
+    >>> a, b = persist(a, b)  # doctest: +SKIP
+
+    Parameters
+    ----------
+    args: Dask collections
+    get : callable, optional
+        A scheduler ``get`` function to use. If not provided, the default
+        is to check the global settings first, and then fall back to
+        the collection defaults.
+    optimize_graph : bool, optional
+        If True [default], the graph is optimized before computation.
+        Otherwise the graph is run as is. This can be useful for debugging.
+    kwargs
+        Extra keywords to forward to the scheduler ``get`` function.
+
+    Returns
+    -------
+    New dask collections backed by in-memory data
+    """
+    collections = [a for a in args if isinstance(a, Base)]
+    if not collections:
+        return args
+
+    try:
+        from distributed.client import default_client
+        collections = default_client().persist(collections, **kwargs)
+        if isinstance(collections, list):  # distributed is inconsistent here
+            collections = tuple(collections)
+        else:
+            collections = (collections,)
+        results_iter = iter(collections)
+        return tuple(a if not isinstance(a, Base)
+                     else next(results_iter)
+                     for a in args)
+
+    except (ImportError, ValueError):
+        pass
+    optimize_graph = kwargs.pop('optimize_graph', True)
+
+    get = kwargs.pop('get', None) or _globals['get']
+
+    if not get:
+        get = collections[0]._default_get
+        if not all(a._default_get == get for a in collections):
+            raise ValueError("Compute called on multiple collections with "
+                             "differing default schedulers. Please specify a "
+                             "scheduler `get` function using either "
+                             "the `get` kwarg or globally with `set_options`.")
+
+    dsk = collections_to_dsk(collections, optimize_graph, **kwargs)
+    keys = list(flatten([var._keys() for var in collections]))
+    results = get(dsk, keys, **kwargs)
+
+    d = dict(zip(keys, results))
+
+    result = [redict_collection(c, {k: d[k]
+                                    for k in flatten(c._keys())})
+              for c in collections]
+    results_iter = iter(result)
+    return tuple(a if not isinstance(a, Base)
+                 else next(results_iter)
+                 for a in args)
diff --git a/dask/tests/test_base.py b/dask/tests/test_base.py
index b25f8df3afe..f6eee8df706 100644
--- a/dask/tests/test_base.py
+++ b/dask/tests/test_base.py
@@ -7,8 +7,10 @@
 import sys
 
 import dask
+from dask import delayed
 from dask.base import (compute, tokenize, normalize_token, normalize_function,
-                       visualize)
+                       visualize, persist)
+from dask.delayed import Delayed
 from dask.utils import tmpdir, tmpfile, ignoring
 from dask.utils_test import inc, dec
 from dask.compatibility import unicode
@@ -259,6 +261,19 @@ def test_compute_array():
     assert np.allclose(out2, arr + 2)
 
 
+@pytest.mark.skipif('not da')
+def test_persist_array():
+    from dask.array.utils import assert_eq
+    arr = np.arange(100).reshape((10, 10))
+    x = da.from_array(arr, chunks=(5, 5))
+    x = (x + 1) - x.mean(axis=0)
+    y = x.persist()
+
+    assert_eq(x, y)
+    assert set(y.dask).issubset(x.dask)
+    assert len(y.dask) == y.npartitions
+
+
 @pytest.mark.skipif('not dd')
 def test_compute_dataframe():
     df = pd.DataFrame({'a': [1, 2, 3, 4], 'b': [5, 5, 3, 3]})
@@ -374,3 +389,42 @@ def test_default_imports():
                  'partd', 's3fs', 'distributed']
     for mod in blacklist:
         assert mod not in modules
+
+
+def test_persist_literals():
+    assert persist(1, 2, 3) == (1, 2, 3)
+
+
+def test_persist_delayed():
+    x1 = delayed(1)
+    x2 = delayed(inc)(x1)
+    x3 = delayed(inc)(x2)
+
+    xx, = persist(x3)
+    assert isinstance(xx, Delayed)
+    assert xx.key == x3.key
+    assert len(xx.dask) == 1
+
+    assert x3.compute() == xx.compute()
+
+
+@pytest.mark.skipif('not da or not db')
+def test_persist_array_bag():
+    x = da.arange(5, chunks=2) + 1
+    b = db.from_sequence([1, 2, 3]).map(inc)
+
+    with pytest.raises(ValueError):
+        persist(x, b)
+
+    xx, bb = persist(x, b, get=dask.async.get_sync)
+
+    assert isinstance(xx, da.Array)
+    assert isinstance(bb, db.Bag)
+
+    assert xx.name == x.name
+    assert bb.name == b.name
+    assert len(xx.dask) == xx.npartitions < len(x.dask)
+    assert len(bb.dask) == bb.npartitions < len(b.dask)
+
+    assert np.allclose(x, xx)
+    assert list(b) == list(bb)
diff --git a/dask/tests/test_distributed.py b/dask/tests/test_distributed.py
index 45b00a0bbf3..5cd3fcc5110 100644
--- a/dask/tests/test_distributed.py
+++ b/dask/tests/test_distributed.py
@@ -1,6 +1,25 @@
 import pytest
 pytest.importorskip('distributed')
 
+from dask import persist, delayed
+from distributed.client import _wait
+from distributed.utils_test import gen_cluster, inc
+
 
 def test_can_import_client():
     from dask.distributed import Client # noqa: F401
+
+
+@gen_cluster(client=True)
+def test_persist(c, s, a, b):
+    x = delayed(inc)(1)
+    x2, = persist(x)
+
+    yield _wait(x2)
+    assert x2.key in a.data or x2.key in b.data
+
+    y = delayed(inc)(10)
+    y2, one = persist(y, 1)
+
+    yield _wait(y2)
+    assert y2.key in a.data or y2.key in b.data


Code review comments:
diff_hunk: @@ -383,3 +387,151 @@ def tokenize(*args, **kwargs):
     if kwargs:
         args = args + (kwargs,)
     return md5(str(tuple(map(normalize_token, args))).encode()).hexdigest()
+
+
+def collections_to_dsk(collections, optimize_graph=True, **kwargs):
+    """
+    Convert many collections into a single dask graph, after optimization
+    """
+    optimizations = (kwargs.pop('optimizations', None) or
+                     _globals.get('optimizations', []))
+    if optimize_graph:
+        groups = groupby(lambda x: x._optimize, collections)
+        groups = {opt: _extract_graph_and_keys(val)
+                  for opt, val in groups.items()}
+        for opt in optimizations:
+            groups = {k: [opt(dsk, keys), keys]
+                      for k, (dsk, keys) in groups.items()}
+        dsk = merge([opt(dsk, keys, **kwargs)
+                     for opt, (dsk, keys) in groups.items()])
+    else:
+        dsk = merge(c.dask for c in collections)
+
+    return dsk
+
+
+def _extract_graph_and_keys(vals):
+    """Given a list of dask vals, return a single graph and a list of keys such
+    that ``get(dsk, keys)`` is equivalent to ``[v.compute() v in vals]``."""
+    dsk = {}
+    keys = []
+    for v in vals:
+        # Optimization to avoid merging dictionaries in Delayed values. Reduces
+        # memory usage for large graphs.
+        if hasattr(v, '_dasks'):
+            for d in v._dasks:
+                dsk.update(d)
+        else:
+            dsk.update(v.dask)
+        keys.append(v._keys())
+
+    return dsk, keys
+
+
+def redict_collection(c, dsk):
+    from dask.delayed import Delayed
+    if isinstance(c, Delayed):
+        return Delayed(c.key, [dsk])
+    else:
+        cc = copy.copy(c)
+        cc.dask = dsk
+        return cc
+
+
+def persist(*args, **kwargs):
+    """ Persist multiple Dask collections into memory
+
+    This turns lazy Dask collections into Dask collections with the same
+    metadata, but now with their results actively computing.
+
+    For example a lazy dask.array built up from many lazy calls will now be a
+    dask.array of the same shape, dtype, etc., but now with all of those lazy
+    operations either in memory (for single-machine computing) or actively
+    computing asynchronously (for distributed computing).
+
+    If using Dask on a single machine then you should ensure that the
+    dataset fits entirely within memory.
+
+    This function operates differently if a ``dask.distributed.Client`` exists
+    and is connected to a distributed scheduler.  In this case this function
+    will return as soon as the task graph has been submitted to the cluster,
+    but before the computations have completed.  Computations will continue
+    asynchronously in the background.  When using this function with the single
+    machine scheduler it blocks until the computations have finished.
+
+    Examples
+    --------
+
+    >>> df = dd.read_csv('/path/to/*.csv')  # doctest: +SKIP
+    >>> df = df[df.name == 'Alice']  # doctest: +SKIP
+    >>> df['in-debt'] = df.balance < 0  # doctest: +SKIP
+    >>> df = df.persist()  # triggers computation  # doctest: +SKIP
+
+    >>> df.value().min()  # future computations are now fast  # doctest: +SKIP
+    -10
+    >>> df.value().max()  # doctest: +SKIP
+    100
+
+    >>> from dask import persist  # use persist function on multiple collections
+    >>> a, b = persist(a, b)  # doctest: +SKIP
+
+    Parameters
+    ----------
+    args: Dask collections
+    get : callable, optional
+        A scheduler ``get`` function to use. If not provided, the default
+        is to check the global settings first, and then fall back to
+        the collection defaults.
+    optimize_graph : bool, optional
+        If True [default], the graph is optimized before computation.
+        Otherwise the graph is run as is. This can be useful for debugging.
+    kwargs
+        Extra keywords to forward to the scheduler ``get`` function.
+
+    Returns
+    -------
+    New dask collections backed by in-memory data
+    """
+    collections = [a for a in args if isinstance(a, Base)]
+    if not collections:
+        return args
+
+    try:
+        from distributed.client import default_client
+        collections = default_client().persist(collections, **kwargs)
+        if isinstance(collections, list):  # distributed is inconsistent here
+            collections = tuple(collections)
+        else:
+            collections = (collections,)
+        results_iter = iter(collections)
+        return tuple(a if not isinstance(a, Base)
+                     else next(results_iter)
+                     for a in args)
+
+    except (ImportError, ValueError):
path: dask/base.py
text: @user1:
This is too much code in a `try` block, especially for exceptions as broad as `ValueError`.

@author:
Done

diff_hunk: @@ -61,8 +62,40 @@ def visualize(self, filename='mydask', format=None, optimize_graph=False,
         return visualize(self, filename=filename, format=format,
                          optimize_graph=optimize_graph, **kwargs)
 
+    def persist(self, **kwargs):
+        """ Persist this dask collection into memory
+
+        This turns a lazy Dask collection into a Dask collection with the same
+        metadata, but now with its results actively computing.  For example a
path: dask/base.py
text: @user1:
computing -> computed or computing

@author:
Resolved

diff_hunk: @@ -61,8 +62,40 @@ def visualize(self, filename='mydask', format=None, optimize_graph=False,
         return visualize(self, filename=filename, format=format,
                          optimize_graph=optimize_graph, **kwargs)
 
+    def persist(self, **kwargs):
+        """ Persist this dask collection into memory
+
+        This turns a lazy Dask collection into a Dask collection with the same
+        metadata, but now with its results actively computing.  For example a
+        lazy dask.array built up from many lazy calls will now be a dask.array
+        of the same shape, dtype, etc., but now with all of those lazy
+        operations either in memory (for single-machine computing) or actively
+        computing asynchronously (for distributed computing).
+
+        If using Dask on a single machine then you should ensure that the
+        dataset fits entirely within memory.
+
+        Parameters
+        ----------
+        get : callable, optional
+            A scheduler ``get`` function to use. If not provided, the default
+            is to check the global settings first, and then fall back to
+            the collection defaults.
+        optimize_graph : bool, optional
+            If True [default], the graph is optimized before computation.
+            Otherwise the graph is run as is. This can be useful for debugging.
+        kwargs
+            Extra keywords to forward to the scheduler ``get`` function.
+        """
+        return persist(self, **kwargs)[0]
path: dask/base.py
text: @user1:
I find tuple unpacking more self-evidently correct, i.e., `result, = persist(self, **kwargs)`

@user2:
I feel the other way - I've been burnt too many times by accidental trailing comas that this initially looks wrong to me.

@user1:
How about using [explicit parentheses](http://stackoverflow.com/questions/3721477/readable-convention-for-unpacking-single-value-tuple) like `(result,) = persist(self, **kwargs)`?


@user2:
That looks better to me. I'm undecided on whether it's clearer than a getitem, but supposedly it's faster :).

@author:
I chose to go with the parentheses solution, which seemed to satisfy both objectives

diff_hunk: @@ -61,8 +62,40 @@ def visualize(self, filename='mydask', format=None, optimize_graph=False,
         return visualize(self, filename=filename, format=format,
                          optimize_graph=optimize_graph, **kwargs)
 
+    def persist(self, **kwargs):
+        """ Persist this dask collection into memory
+
+        This turns a lazy Dask collection into a Dask collection with the same
+        metadata, but now with its results actively computing.  For example a
+        lazy dask.array built up from many lazy calls will now be a dask.array
+        of the same shape, dtype, etc., but now with all of those lazy
+        operations either in memory (for single-machine computing) or actively
path: dask/base.py
text: @user1:
"lazy operations in memory" doesn't really make sense. Maybe talk about "the result of those lazy operations", or just say the array is stored in memory as a NumPy array?

@author:
Resolved

diff_hunk: @@ -498,18 +481,24 @@ def persist(*args, **kwargs):
 
     try:
         from distributed.client import default_client
-        collections = default_client().persist(collections, **kwargs)
-        if isinstance(collections, list):  # distributed is inconsistent here
-            collections = tuple(collections)
+    except ImportError:
+        pass
+    else:
+        try:
+            client = default_client()
+        except ValueError:
path: dask/base.py
text: @user1:
This is fine, but going forward you might consider refactoring `default_client` to either return a sentinel value (i.e., `None`) if there is no default or raise a sentinel specific error, e.g., `NoDefaultClientError`.

@author:
Yes, having gone through this experience I now agree.  However I've also learned to avoid changing both libraries simultaneously if I can avoid it :)