Add purge utility for query caching and add test.

Author: guzman-raphael

datajoint/connection.py

@@ -228,13 +228,14 @@ def set_query_cache(self, query_cache=None):
         """
         self._query_cache = query_cache
 
-    def purge_query_cache(self, query_cache):
-        """
-        Purges if query cache is available with the provided reference.
-
-        :param query_cache: a string associated with the hash for query results
-        """
-        pass  # wip
+    def purge_query_cache(self):
+        """ Purges all query cache. """
+        if 'query_cache' in config and isinstance(config['query_cache'], str) and \
+                pathlib.Path(config['query_cache']).is_dir():
+            path_iter = pathlib.Path(config['query_cache']).glob('**/*')
+            for path in path_iter:
+                path.unlink()
+        self._query_cache = None
 
     def close(self):
         self._conn.close()

docs-parts/queries/12-Query-Caching_lang3.rst

@@ -1,6 +1,6 @@
 
 .. code-block:: python
 
-  # deactivate query caching
-  conn.purge_query_cache(query_cache='main')
+  # purged the cached queries
+  conn.purge_query_cache()
 

tests/test_fetch.py

@@ -7,6 +7,7 @@
 import warnings
 from . import schema
 import datajoint as dj
+import os
 
 
 class TestFetch:
@@ -254,3 +255,34 @@ def test_same_secondary_attribute(self):
         children = (schema.Child * schema.Parent().proj()).fetch()['name']
         assert len(children) == 1
         assert children[0] == 'Dan'
+
+    def test_query_caching(self):
+        # initialize cache directory
+        os.mkdir(os.path.expanduser('~/dj_query_cache'))
+
+        with dj.config(query_cache=os.path.expanduser('~/dj_query_cache')):
+            # insert sample data and load cache
+            schema.TTest3.insert([dict(key=100+i, value=200+i) for i in range(2)])
+            dj.conn().set_query_cache(query_cache='main')
+            cached_res = schema.TTest3().fetch()
+            # attempt to insert while caching enabled
+            try:
+                schema.TTest3.insert([dict(key=200+i, value=400+i) for i in range(2)])
+                assert False, 'Insert allowed which query caching enabled'
+            except dj.DataJointError:
+                dj.conn().set_query_cache()
+            # insert new data
+            schema.TTest3.insert([dict(key=600+i, value=800+i) for i in range(2)])
+            # re-enable cache to access old results
+            dj.conn().set_query_cache(query_cache='main')
+            previous_cache = schema.TTest3().fetch()
+            # verify properly cached and how to refresh results
+            assert all([c == p for c, p in zip(cached_res, previous_cache)])
+            dj.conn().set_query_cache()
+            uncached_res = schema.TTest3().fetch()
+            assert len(uncached_res) > len(cached_res)
+            # purge query cache
+            dj.conn().purge_query_cache()
+
+        # reset cache directory state (will fail if purge was unsuccessful)
+        os.rmdir(os.path.expanduser('~/dj_query_cache'))