Merge pull request pandas-dev#60 from manahl/write_transation_without_audit

Add support non-audited transactions.

Author: jamesblackburn

arctic/store/audit.py

@@ -31,7 +31,7 @@ class ArcticTransaction(object):
     call the `write` method of the context manager to output changes. The changes will only be written when the block
     exits.
 
-    NB changes are audited.
+    NB changes may be audited.
 
     Example:
     -------
@@ -44,7 +44,8 @@ class ArcticTransaction(object):
     retry the whole block should that happens, as the assumption is that you need to base your changes on a different
     starting timeseries.
     '''
-    def __init__(self, version_store, symbol, user, log, modify_timeseries=None, *args, **kwargs):
+    def __init__(self, version_store, symbol, user, log, modify_timeseries=None, audit=True,
+                 *args, **kwargs):
         '''
         Parameters
         ----------
@@ -67,13 +68,20 @@ def __init__(self, version_store, symbol, user, log, modify_timeseries=None, *ar
             interacting with code that read in the data already and for some reason you cannot refactor the read-write
             operation to be contained within this context manager
 
+        audit: `bool`
+            should we 'audit' the transaction. An audited write transaction is equivalent to a snapshot
+            before and after the data change - i.e. we won't prune versions of the data involved in an
+            audited transaction.  This can be used to ensure that the history of certain data changes is
+            preserved indefinitely.
+
         all other args:
             Will be passed into the initial read
         '''
         self._version_store = version_store
         self._symbol = symbol
         self._user = user
         self._log = log
+        self._audit = audit
         logger.info("MT: {}@{}: [{}] {}: {}".format(_get_host(version_store).get('l'),
                                                     _get_host(version_store).get('mhost'),
                                                        user, log, symbol)
@@ -136,4 +144,5 @@ def __exit__(self, *args, **kwargs):
                                                       self._symbol, self.base_ts.version, written_ver.version))
 
             changed = ChangedItem(self._symbol, self.base_ts, written_ver, None)
-            self._version_store._write_audit(self._user, self._log, changed)
+            if self._audit:
+                self._version_store._write_audit(self._user, self._log, changed)

tests/unit/store/test_version_store_audit.py

@@ -8,7 +8,7 @@
 from arctic.exceptions import ConcurrentModificationException, NoDataFoundException
 
 
-def test_ConcurrentWriteBlock_simple():
+def test_ArcticTransaction_simple():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
@@ -20,11 +20,27 @@ def test_ConcurrentWriteBlock_simple():
         cwb.write(sentinel.symbol, pd.DataFrame(index=[3, 4], data={'a': [1.0, 2.0]}), metadata=sentinel.meta)
 
     assert not vs._delete_version.called
-    vs.write.assert_called_once_with(sentinel.symbol, ANY, prune_previous_version=True, metadata=sentinel.meta)
-    vs.list_versions.assert_called_once_with(sentinel.symbol)
+    assert vs.write.call_args_list == [call(sentinel.symbol, ANY, prune_previous_version=True, metadata=sentinel.meta)]
+    assert vs.list_versions.call_args_list == [call(sentinel.symbol)]
+    assert vs._write_audit.call_args_list == [call(sentinel.user, sentinel.log, ANY)]
+
+
+def test_ArticTransaction_no_audit():
+    vs = create_autospec(VersionStore, _collection=Mock())
+    ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
+    vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
+    vs.write.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=2,
+                            metadata=None, data=None)
+    vs.list_versions.return_value = [{'version': 2}, {'version': 1}]
+
+    with ArcticTransaction(vs, sentinel.symbol, sentinel.user, sentinel.log, audit=False) as cwb:
+        cwb.write(sentinel.symbol, pd.DataFrame(index=[3, 4], data={'a': [1.0, 2.0]}), metadata=sentinel.meta)
+
+    assert vs.write.call_count == 1
+    assert vs._write_audit.call_count == 0
 
 
-def test_ConcurrentWriteBlock_writes_if_metadata_changed():
+def test_ArcticTransaction_writes_if_metadata_changed():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
@@ -49,7 +65,7 @@ def test_ConcurrentWriteBlock_writes_if_metadata_changed():
         assert cwb._do_write is False
 
 
-def test_ConcurrentWriteBlock_writes_if_base_data_corrupted():
+def test_ArcticTransaction_writes_if_base_data_corrupted():
 
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
@@ -67,7 +83,7 @@ def test_ConcurrentWriteBlock_writes_if_base_data_corrupted():
     assert vs.list_versions.call_args_list == [call(sentinel.symbol)]
 
 
-def test_ConcurrentWriteBlock_writes_no_data_found():
+def test_ArcticTransaction_writes_no_data_found():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.side_effect = NoDataFoundException('no data')
@@ -85,7 +101,7 @@ def test_ConcurrentWriteBlock_writes_no_data_found():
                                               call(sentinel.symbol)]
 
 
-def test_ConcurrentWriteBlock_writes_no_data_found_deleted():
+def test_ArcticTransaction_writes_no_data_found_deleted():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.side_effect = NoDataFoundException('no data')
@@ -103,7 +119,7 @@ def test_ConcurrentWriteBlock_writes_no_data_found_deleted():
                                               call(sentinel.symbol)]
 
 
-def test_ConcurrentWriteBlock_does_nothing_when_data_not_modified():
+def test_ArcticTransaction_does_nothing_when_data_not_modified():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
@@ -117,7 +133,7 @@ def test_ConcurrentWriteBlock_does_nothing_when_data_not_modified():
     assert not vs.write.called
 
 
-def test_ConcurrentWriteBlock_does_nothing_when_data_is_None():
+def test_ArcticTransaction_does_nothing_when_data_is_None():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
@@ -131,7 +147,7 @@ def test_ConcurrentWriteBlock_does_nothing_when_data_is_None():
     assert not vs.write.called
 
 
-def test_ConcurrentWriteBlock_guards_against_inconsistent_ts():
+def test_ArcticTransaction_guards_against_inconsistent_ts():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)
@@ -144,7 +160,7 @@ def test_ConcurrentWriteBlock_guards_against_inconsistent_ts():
             pass
 
 
-def test_ConcurrentWriteBlock_detects_concurrent_writes():
+def test_ArcticTransaction_detects_concurrent_writes():
     vs = create_autospec(VersionStore, _collection=Mock())
     ts1 = pd.DataFrame(index=[1, 2], data={'a':[1.0, 2.0]})
     vs.read.return_value = VersionedItem(symbol=sentinel.symbol, library=sentinel.library, version=1, metadata=None, data=ts1)