Add copy() method to client.Database (issue 74).

cmlenz · cmlenz · commit d2eb5b12971a · 2009-06-29T13:11:29.000Z
--HG--
extra : convert_revision : svn%3A7a298fb0-333a-0410-83e7-658617cd9cf3/trunk%40155
diff --git a/ChangeLog.txt b/ChangeLog.txt
@@ -18,6 +18,8 @@ http://couchdb-python.googlecode.com/svn/tags/0.6.0
  * Removed the `eager` option on the `query()` and `view()` methods of
    `schema.Document`. Use the `include_docs` option instead, which doesn't
    require an additional request per document.
+ * Added a `copy()` method to the `client.Database` class, which translates to
+   a HTTP COPY request (issue 74).
 
 
 Version 0.5
diff --git a/couchdb/client.py b/couchdb/client.py
@@ -346,6 +346,43 @@ def compact(self):
         resp, data = self.resource.post('_compact')
         return data['ok']
 
+    def copy(self, src, dest):
+        """Copy the given document to create a new document.
+
+        :param src: the ID of the document to copy, or a dictionary or
+                    `Document` object representing the source document.
+        :param dest: either the destination document ID as string, or a
+                     dictionary or `Document` instance of the document that
+                     should be overwritten.
+        :return: the new revision of the destination document
+        :rtype: `str`
+        """
+        if not isinstance(src, basestring):
+            if not isinstance(src, dict):
+                if hasattr(src, 'items'):
+                    src = src.items()
+                else:
+                    raise TypeError('expected dict or string, got %s' %
+                                    type(src))
+            src = src['_id']
+
+        if not isinstance(dest, basestring):
+            if not isinstance(dest, dict):
+                if hasattr(dest, 'items'):
+                    dest = src.items()
+                else:
+                    raise TypeError('expected dict or string, got %s' %
+                                    type(dest))
+            if '_rev' in dest:
+                dest = '%s?rev=%s' % (dest['_id'], dest['_rev'])
+            else:
+                dest = dest['_id']
+
+        resp, data = self.resource._request('COPY', src,
+                                            headers={'Destination': dest})
+        return data['rev']
+
+
     def delete(self, doc):
         """Delete the given document from the database.
 
diff --git a/couchdb/tests/client.py b/couchdb/tests/client.py
@@ -223,6 +223,29 @@ def test_bulk_update_all_or_nothing(self):
         doc = self.db.get(doc['_id'], conflicts=True)
         assert '_conflicts' in doc
 
+    def test_copy_doc(self):
+        self.db['foo'] = {'status': 'testing'}
+        result = self.db.copy('foo', 'bar')
+        self.assertEqual(result, self.db['bar'].rev)
+
+    def test_copy_doc_conflict(self):
+        self.db['bar'] = {'status': 'idle'}
+        self.db['foo'] = {'status': 'testing'}
+        self.assertRaises(client.ResourceConflict, self.db.copy, 'foo', 'bar')
+
+    def test_copy_doc_overwrite(self):
+        self.db['bar'] = {'status': 'idle'}
+        self.db['foo'] = {'status': 'testing'}
+        result = self.db.copy('foo', self.db['bar'])
+        doc = self.db['bar']
+        self.assertEqual(result, doc.rev)
+        self.assertEqual('testing', doc['status'])
+
+    def test_copy_doc_srcobj(self):
+        self.db['foo'] = {'status': 'testing'}
+        self.db.copy(self.db['foo'], 'bar')
+        self.assertEqual('testing', self.db['bar']['status'])
+
 
 def suite():
     suite = unittest.TestSuite()
