diff --git a/oracle/soda.py b/oracle/soda.py index 8010b2e..ef66cb5 100644 --- a/oracle/soda.py +++ b/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) diff --git a/tests/test_integration.py b/tests/test_integration.py index 41583f5..9b154a1 100644 --- a/tests/test_integration.py +++ b/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())