SODA: update, delete, and bulk insert (#202)

oracle/soda.py

@@ -98,6 +98,33 @@
 # getDocuments() materialises into host arrays, so it needs a capacity. With no
 # limit set, cap at this many and raise on overflow rather than truncate.
 _DEFAULT_FETCH_CAP = 1000
+# Update / delete / bulk (#202). replace_one returns a NUMBER (1 replaced / 0
+# not); replace_one_and_get returns the new SODA_DOCUMENT_T (NULL if nothing
+# matched); remove returns the deleted count. DBMS_SODA's bulk insert_many and
+# save are ORA-03001 "unimplemented feature" in thin mode, so insertMany loops
+# insert_one over a host array of contents instead.
+_REPLACE_ONE = (
+    "DECLARE c SODA_COLLECTION_T; op SODA_OPERATION_T; d SODA_DOCUMENT_T; "
+    "res NUMBER; BEGIN c := DBMS_SODA.open_collection(:name); " + _OP_BUILD +
+    "d := SODA_DOCUMENT_T(b_content => :content); res := op.replace_one(d); "
+    ":replaced := res; END;")
+_REPLACE_ONE_AND_GET = (
+    "DECLARE c SODA_COLLECTION_T; op SODA_OPERATION_T; d SODA_DOCUMENT_T; "
+    "r SODA_DOCUMENT_T; BEGIN c := DBMS_SODA.open_collection(:name); " + _OP_BUILD +
+    "d := SODA_DOCUMENT_T(b_content => :content); r := op.replace_one_and_get(d); "
+    "IF r IS NULL THEN :missing := 1; ELSE :missing := 0; "
+    ":rkey := r.get_key(); :rver := r.get_version(); :rmt := r.get_media_type(); "
+    ":rcreated := r.get_created_on; :rmodified := r.get_last_modified; "
+    "END IF; END;")
+_REMOVE = (
+    "DECLARE c SODA_COLLECTION_T; op SODA_OPERATION_T; BEGIN "
+    "c := DBMS_SODA.open_collection(:name); " + _OP_BUILD +
+    ":n := op.remove(); END;")
+_INSERT_MANY = (
+    "DECLARE c SODA_COLLECTION_T; d SODA_DOCUMENT_T; n NUMBER; BEGIN "
+    "c := DBMS_SODA.open_collection(:name); "
+    "FOR i IN 1..:cnt LOOP d := SODA_DOCUMENT_T(b_content => :contents(i)); "
+    "n := c.insert_one(d); END LOOP; END;")
 
 
 def _names_query(start_name, limit):
@@ -282,6 +309,40 @@ def getDocuments(self) -> "list[SodaDocument]":
         _check_fetch_overflow(b, cap)
         return _docs_from_arrays(b)
 
+    def replaceOne(self, doc) -> bool:
+        """Replace the matched document's content. Returns True if one was
+        replaced, False if nothing matched."""
+        _, content, _ = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        replaced = cur.var(DB_TYPE_NUMBER)
+        b = self._in_binds()
+        b.update({"content": content, "replaced": replaced})
+        cur.execute(_REPLACE_ONE, b)
+        return bool(replaced.getvalue())
+
+    def replaceOneAndGet(self, doc) -> "SodaDocument | None":
+        """Replace the matched document and return a `SodaDocument` with the new
+        key / version / metadata (no content), or None if nothing matched."""
+        _, content, _ = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur, content=False)
+        b["missing"] = cur.var(DB_TYPE_NUMBER)
+        b.update(self._in_binds())
+        b["content"] = content
+        cur.execute(_REPLACE_ONE_AND_GET, b)
+        if b["missing"].getvalue():
+            return None
+        return _doc_from_binds(b, with_content=False)
+
+    def remove(self) -> int:
+        """Remove the matched documents; returns the number removed."""
+        cur = self._connection.cursor()
+        n = cur.var(DB_TYPE_NUMBER)
+        b = self._in_binds()
+        b["n"] = n
+        cur.execute(_REMOVE, b)
+        return int(n.getvalue())
+
 
 class SodaCollection:
     """A SODA collection (#163). Obtain one from `SodaDatabase.createCollection`
@@ -338,6 +399,18 @@ def insertOneAndGet(self, doc) -> SodaDocument:
         cur.execute(_INSERT_ONE_AND_GET, b)
         return _doc_from_binds(b, with_content=False)
 
+    def insertMany(self, docs) -> None:
+        """Insert several documents (each a `SodaDocument` or a bare value) in a
+        single round trip. Each document's content is subject to the 32767-byte
+        inline limit (#200)."""
+        contents = [_doc_to_bind(d)[1] for d in docs]
+        if not contents:
+            return
+        cur = self._connection.cursor()
+        arr = cur.arrayvar(DB_TYPE_RAW, contents)
+        cur.execute(_INSERT_MANY,
+                    {"name": self.name, "cnt": len(contents), "contents": arr})
+
     def find(self) -> SodaOperation:
         """Begin a read operation (chain `.key(k).getOne()`)."""
         return SodaOperation(self)
@@ -517,6 +590,35 @@ async def getDocuments(self) -> "list[SodaDocument]":
         _check_fetch_overflow(b, cap)
         return _docs_from_arrays(b)
 
+    async def replaceOne(self, doc) -> bool:
+        _, content, _ = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        replaced = cur.var(DB_TYPE_NUMBER)
+        b = self._in_binds()
+        b.update({"content": content, "replaced": replaced})
+        await cur.execute(_REPLACE_ONE, b)
+        return bool(replaced.getvalue())
+
+    async def replaceOneAndGet(self, doc) -> "SodaDocument | None":
+        _, content, _ = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur, content=False)
+        b["missing"] = cur.var(DB_TYPE_NUMBER)
+        b.update(self._in_binds())
+        b["content"] = content
+        await cur.execute(_REPLACE_ONE_AND_GET, b)
+        if b["missing"].getvalue():
+            return None
+        return _doc_from_binds(b, with_content=False)
+
+    async def remove(self) -> int:
+        cur = self._connection.cursor()
+        n = cur.var(DB_TYPE_NUMBER)
+        b = self._in_binds()
+        b["n"] = n
+        await cur.execute(_REMOVE, b)
+        return int(n.getvalue())
+
 
 class AsyncSodaCollection:
     """Async counterpart to `SodaCollection` (#163 / #200)."""
@@ -565,6 +667,16 @@ async def insertOneAndGet(self, doc) -> SodaDocument:
         await cur.execute(_INSERT_ONE_AND_GET, b)
         return _doc_from_binds(b, with_content=False)
 
+    async def insertMany(self, docs) -> None:
+        contents = [_doc_to_bind(d)[1] for d in docs]
+        if not contents:
+            return
+        cur = self._connection.cursor()
+        arr = cur.arrayvar(DB_TYPE_RAW, contents)
+        await cur.execute(_INSERT_MANY,
+                          {"name": self.name, "cnt": len(contents),
+                           "contents": arr})
+
     def find(self) -> AsyncSodaOperation:
         return AsyncSodaOperation(self)
 

tests/test_integration.py

@@ -2323,6 +2323,34 @@ def test_getdocuments_overflow_guard(self):
         finally:
             _soda._DEFAULT_FETCH_CAP = saved
 
+    def test_update_delete_bulk(self):
+        col = self.soda.createCollection("it_udb")
+        # insertMany (one round trip)
+        col.insertMany([{"n": 1}, {"n": 2}, {"n": 3}])
+        self.assertEqual(col.find().count(), 3)
+        saved = col.insertOneAndGet({"v": 1})
+
+        # replaceOne by key (match + no match)
+        self.assertTrue(col.find().key(saved.key).replaceOne({"v": 2}))
+        self.assertEqual(col.find().key(saved.key).getOne().getContent()["v"], 2)
+        self.assertFalse(
+            col.find().key("AABBCCDDEEFF00112233445566").replaceOne({"v": 9}))
+
+        # replaceOneAndGet -> new version; None when nothing matches
+        got = col.find().key(saved.key).replaceOneAndGet({"v": 3})
+        self.assertEqual(got.key, saved.key)
+        self.assertTrue(got.version)
+        self.assertIsNone(
+            col.find().key("AABBCCDDEEFF00112233445566").replaceOneAndGet({}))
+
+        # remove by filter -> count removed
+        self.assertEqual(col.find().filter({"v": {"$exists": True}}).remove(), 1)
+        self.assertEqual(col.find().count(), 3)
+        # remove by key
+        k = col.find().limit(1).getDocuments()[0].key
+        self.assertEqual(col.find().key(k).remove(), 1)
+        self.assertEqual(col.find().count(), 2)
+
 
 @unittest.skipUnless(_USER, _SKIP_REASON)
 class SqlDomainIntegration(_IntegrationBase):
@@ -3224,6 +3252,12 @@ async def test_async_soda_collections(self):
             docs = await col.find().filter({"x": 2}).getDocuments()
             self.assertEqual([d.getContent()["x"] for d in docs], [2])
             self.assertIsNone(await col.find().filter({"x": 999}).getOne())
+            # update / delete / bulk on the async path (#202)
+            await col.insertMany([{"x": 3}, {"x": 4}])
+            self.assertTrue(await col.find().key(saved.key).replaceOne({"x": 5}))
+            g = await col.find().key(saved.key).replaceOneAndGet({"x": 6})
+            self.assertEqual(g.key, saved.key)
+            self.assertEqual(await col.find().key(saved.key).remove(), 1)
             self.assertTrue(await col.drop())
             self.assertFalse(await col.drop())