SODA: document model + insert + read by key (#200)

oracle/soda.py

@@ -8,22 +8,35 @@
 to speak; the documented PL/SQL API is the thin path and pyoracle already drives
 PL/SQL blocks + binds. The public surface mirrors oracledb's SODA API (camelCase,
 following the cross-language SODA spec): ``connection.getSodaDatabase()`` returns
-a `SodaDatabase`, whose collections are `SodaCollection` objects.
+a `SodaDatabase`, whose collections are `SodaCollection` objects holding
+`SodaDocument` objects.
 
 SODA needs an Oracle 18c+ server (``DBMS_SODA``); `getSodaDatabase` raises
-``NotSupportedError`` below that. This module is collection management (#199);
-documents / QBE / updates land in the follow-up sub-tickets of #163.
+``NotSupportedError`` below that. This module covers collection management (#199)
+and the document model — insert + read by key (#200). QBE find, update and
+delete land in the follow-up sub-tickets of #163.
+
+A document's content is bound into ``DBMS_SODA`` as bytes; a value over the
+32767-byte PL/SQL limit rides pyoracle's transparent temp-LOB bind (#91), so
+*inserting* large documents works. *Reading* content back comes through a single
+``DBMS_LOB.SUBSTR`` and is therefore capped at 32767 bytes — a document whose
+stored content exceeds that raises rather than silently truncating; chunked
+large-content reads are a follow-up.
 """
 
 import json
 
-from oracle.datatypes import DB_TYPE_NUMBER, DB_TYPE_VARCHAR
+from oracle.datatypes import DB_TYPE_NUMBER, DB_TYPE_RAW, DB_TYPE_VARCHAR
 from oracle.exceptions import NotSupportedError
 
 # DBMS_SODA / SODA shipped in Oracle 18c (server major version 18).
 _SODA_MIN_MAJOR = 18
+# A single DBMS_LOB.SUBSTR / cast_to_raw read tops out at the 32767-byte PL/SQL
+# VARCHAR2/RAW limit.
+_MAX_INLINE = 32767
+_DEFAULT_MEDIA_TYPE = "application/json"
 
-# PL/SQL templates shared by the sync and async collection/database classes.
+# --- PL/SQL templates (shared by the sync and async classes) ---------------
 _CREATE = ("DECLARE c SODA_COLLECTION_T; BEGIN "
            "c := DBMS_SODA.create_collection(:name); END;")
 _CREATE_MD = ("DECLARE c SODA_COLLECTION_T; BEGIN "
@@ -37,6 +50,28 @@
              ":md := c.get_metadata(); END;")
 _TRUNCATE = ("DECLARE c SODA_COLLECTION_T; n NUMBER; BEGIN "
              "c := DBMS_SODA.open_collection(:name); n := c.truncate(); END;")
+_INSERT_ONE = (
+    "DECLARE c SODA_COLLECTION_T; d SODA_DOCUMENT_T; n NUMBER; BEGIN "
+    "c := DBMS_SODA.open_collection(:name); "
+    "d := SODA_DOCUMENT_T(key => :key, b_content => :content, "
+    "media_type => :mt); n := c.insert_one(d); END;")
+_INSERT_ONE_AND_GET = (
+    "DECLARE c SODA_COLLECTION_T; d SODA_DOCUMENT_T; r SODA_DOCUMENT_T; BEGIN "
+    "c := DBMS_SODA.open_collection(:name); "
+    "d := SODA_DOCUMENT_T(key => :key, b_content => :content, "
+    "media_type => :mt); r := c.insert_one_and_get(d); "
+    ":rkey := r.get_key(); :rver := r.get_version(); :rmt := r.get_media_type();"
+    " :rcreated := r.get_created_on; :rmodified := r.get_last_modified; END;")
+_GET_ONE = (
+    "DECLARE c SODA_COLLECTION_T; d SODA_DOCUMENT_T; b BLOB; BEGIN "
+    "c := DBMS_SODA.open_collection(:name); "
+    "d := c.find().key(:key).get_one(); "
+    "IF d IS NULL THEN :missing := 1; ELSE :missing := 0; "
+    "b := d.get_blob(); :clen := DBMS_LOB.GETLENGTH(b); "
+    ":content := DBMS_LOB.SUBSTR(b, 32767, 1); "
+    ":rkey := d.get_key(); :rver := d.get_version(); :rmt := d.get_media_type();"
+    " :rcreated := d.get_created_on; :rmodified := d.get_last_modified; "
+    "END IF; END;")
 
 
 def _names_query(start_name, limit):
@@ -71,14 +106,110 @@ def _norm_metadata(metadata):
     return metadata
 
 
+def _encode_content(content) -> bytes:
+    # A document's content as UTF-8 bytes for binding: dict/list -> JSON text,
+    # str -> encoded, bytes -> as-is.
+    if isinstance(content, (bytes, bytearray)):
+        return bytes(content)
+    if isinstance(content, str):
+        return content.encode("utf-8")
+    return json.dumps(content).encode("utf-8")
+
+
+class SodaDocument:
+    """A SODA document (#200). Either built client-side by
+    `SodaDatabase.createDocument` (content set; no key/version/timestamps until
+    inserted) or returned by an insert / read (server metadata populated).
+
+    `content` is held as raw bytes; `getContent()` parses it as JSON for a JSON
+    media type, `getContentAsString()` / `getContentAsBytes()` return it as
+    text / bytes."""
+
+    def __init__(self, content=None, key=None, version=None,
+                 mediaType=_DEFAULT_MEDIA_TYPE, createdOn=None,
+                 lastModified=None):
+        self._content = content            # bytes or None
+        self.key = key
+        self.version = version
+        self.mediaType = mediaType
+        self.createdOn = createdOn
+        self.lastModified = lastModified
+
+    def __repr__(self) -> str:
+        return f"SodaDocument(key={self.key!r}, mediaType={self.mediaType!r})"
+
+    def getContentAsBytes(self) -> bytes | None:
+        return self._content
+
+    def getContentAsString(self, encoding: str = "utf-8") -> str | None:
+        if self._content is None:
+            return None
+        return self._content.decode(encoding)
+
+    def getContent(self):
+        """The parsed content: a Python value for a JSON document, otherwise the
+        decoded string."""
+        if self._content is None:
+            return None
+        if self.mediaType == _DEFAULT_MEDIA_TYPE:
+            return json.loads(self._content)
+        return self.getContentAsString()
+
+
+def _doc_to_bind(doc) -> tuple:
+    # A SodaDocument or a bare content value -> (key, content_bytes, mediaType)
+    # for the insert templates.
+    if isinstance(doc, SodaDocument):
+        return doc.key, _encode_content(doc._content), doc.mediaType
+    return None, _encode_content(doc), _DEFAULT_MEDIA_TYPE
+
+
+def _content_or_raise(content, length):
+    # Guard against a silently truncated read: the single SUBSTR maxes at
+    # _MAX_INLINE bytes (#200 limitation).
+    if length is not None and int(length) > _MAX_INLINE:
+        raise NotSupportedError(
+            f"SODA document content is {int(length)} bytes; reading content "
+            f"over {_MAX_INLINE} bytes is not yet supported")
+    if content is None:
+        return None
+    return bytes(content)
+
+
 # --------------------------------------------------------------------------
 # Sync
 # --------------------------------------------------------------------------
 
+class SodaOperation:
+    """A SODA read operation builder (#200): `collection.find()` returns one;
+    chain a terminal like `.key(k).getOne()`. QBE filters / counts / multi-doc
+    reads extend this in the #163 follow-ups."""
+
+    def __init__(self, collection: "SodaCollection"):
+        self._collection = collection
+        self._connection = collection._connection
+        self._key = None
+
+    def key(self, value: str) -> "SodaOperation":
+        self._key = value
+        return self
+
+    def getOne(self) -> "SodaDocument | None":
+        """The single document matched by `.key(...)`, or None if there is
+        none."""
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur)
+        b.update({"name": self._collection.name, "key": self._key})
+        cur.execute(_GET_ONE, b)
+        if b["missing"].getvalue():
+            return None
+        return _doc_from_binds(b, with_content=True)
+
+
 class SodaCollection:
     """A SODA collection (#163). Obtain one from `SodaDatabase.createCollection`
-    / `openCollection`. This step covers name / metadata / drop / truncate;
-    document operations arrive in the #163 follow-ups."""
+    / `openCollection`. Covers name / metadata / drop / truncate (#199) and
+    document insert + read by key (#200)."""
 
     def __init__(self, database: "SodaDatabase", name: str, metadata=None):
         self._database = database
@@ -112,6 +243,28 @@ def truncate(self) -> None:
         cur = self._connection.cursor()
         cur.execute(_TRUNCATE, {"name": self.name})
 
+    def insertOne(self, doc) -> None:
+        """Insert a document (a `SodaDocument` or a bare content value)."""
+        key, content, mt = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        cur.execute(_INSERT_ONE,
+                    {"name": self.name, "key": key, "content": content,
+                     "mt": mt})
+
+    def insertOneAndGet(self, doc) -> SodaDocument:
+        """Insert a document and return a `SodaDocument` carrying the resulting
+        key / version / metadata (no content, matching oracledb)."""
+        key, content, mt = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur, content=False)
+        b.update({"name": self.name, "key": key, "content": content, "mt": mt})
+        cur.execute(_INSERT_ONE_AND_GET, b)
+        return _doc_from_binds(b, with_content=False)
+
+    def find(self) -> SodaOperation:
+        """Begin a read operation (chain `.key(k).getOne()`)."""
+        return SodaOperation(self)
+
 
 class SodaDatabase:
     """The SODA entry point for a connection (oracledb's SodaDatabase, #163).
@@ -149,13 +302,70 @@ def getCollectionNames(self, startName: str | None = None,
         cur.execute(sql, binds)
         return [Row[0] for Row in cur.fetchall()]
 
+    def createDocument(self, content, key: str | None = None,
+                       mediaType: str = _DEFAULT_MEDIA_TYPE) -> SodaDocument:
+        """Build a client-side `SodaDocument` from `content` (a dict / str /
+        bytes), ready to insert."""
+        return SodaDocument(content=_encode_content(content), key=key,
+                            mediaType=mediaType)
+
+
+# --- helpers shared by the document-returning operations -------------------
+
+def _new_doc_out_binds(cur, content: bool = True) -> dict:
+    # The OUT binds the insert/read templates fill: key, version, media type,
+    # created/last-modified timestamps, and (for a read) the content + length.
+    binds = {"rkey": cur.var(DB_TYPE_VARCHAR), "rver": cur.var(DB_TYPE_VARCHAR),
+             "rmt": cur.var(DB_TYPE_VARCHAR),
+             "rcreated": cur.var(DB_TYPE_VARCHAR),
+             "rmodified": cur.var(DB_TYPE_VARCHAR)}
+    if content:
+        binds["content"] = cur.var(DB_TYPE_RAW)
+        binds["clen"] = cur.var(DB_TYPE_NUMBER)
+        binds["missing"] = cur.var(DB_TYPE_NUMBER)
+    return binds
+
+
+def _doc_from_binds(b: dict, with_content: bool) -> SodaDocument:
+    content = None
+    if with_content:
+        content = _content_or_raise(b["content"].getvalue(),
+                                    b["clen"].getvalue())
+    return SodaDocument(
+        content=content, key=b["rkey"].getvalue(), version=b["rver"].getvalue(),
+        mediaType=b["rmt"].getvalue() or _DEFAULT_MEDIA_TYPE,
+        createdOn=b["rcreated"].getvalue(),
+        lastModified=b["rmodified"].getvalue())
+
 
 # --------------------------------------------------------------------------
 # Async (mirrors the sync surface; same PL/SQL, awaited)
 # --------------------------------------------------------------------------
 
+class AsyncSodaOperation:
+    """Async counterpart to `SodaOperation` (#200)."""
+
+    def __init__(self, collection: "AsyncSodaCollection"):
+        self._collection = collection
+        self._connection = collection._connection
+        self._key = None
+
+    def key(self, value: str) -> "AsyncSodaOperation":
+        self._key = value
+        return self
+
+    async def getOne(self) -> "SodaDocument | None":
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur)
+        b.update({"name": self._collection.name, "key": self._key})
+        await cur.execute(_GET_ONE, b)
+        if b["missing"].getvalue():
+            return None
+        return _doc_from_binds(b, with_content=True)
+
+
 class AsyncSodaCollection:
-    """Async counterpart to `SodaCollection` (#163)."""
+    """Async counterpart to `SodaCollection` (#163 / #200)."""
 
     def __init__(self, database: "AsyncSodaDatabase", name: str, metadata=None):
         self._database = database
@@ -186,6 +396,24 @@ async def truncate(self) -> None:
         cur = self._connection.cursor()
         await cur.execute(_TRUNCATE, {"name": self.name})
 
+    async def insertOne(self, doc) -> None:
+        key, content, mt = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        await cur.execute(_INSERT_ONE,
+                          {"name": self.name, "key": key, "content": content,
+                           "mt": mt})
+
+    async def insertOneAndGet(self, doc) -> SodaDocument:
+        key, content, mt = _doc_to_bind(doc)
+        cur = self._connection.cursor()
+        b = _new_doc_out_binds(cur, content=False)
+        b.update({"name": self.name, "key": key, "content": content, "mt": mt})
+        await cur.execute(_INSERT_ONE_AND_GET, b)
+        return _doc_from_binds(b, with_content=False)
+
+    def find(self) -> AsyncSodaOperation:
+        return AsyncSodaOperation(self)
+
 
 class AsyncSodaDatabase:
     """Async counterpart to `SodaDatabase` (#163)."""
@@ -217,3 +445,8 @@ async def getCollectionNames(self, startName: str | None = None,
         sql, binds = _names_query(startName, limit)
         await cur.execute(sql, binds)
         return [Row[0] for Row in await cur.fetchall()]
+
+    def createDocument(self, content, key: str | None = None,
+                       mediaType: str = _DEFAULT_MEDIA_TYPE) -> SodaDocument:
+        return SodaDocument(content=_encode_content(content), key=key,
+                            mediaType=mediaType)

tests/test_integration.py

@@ -2262,6 +2262,36 @@ def test_truncate(self):
             ":1 := c.find().count(); END;", [v])
         self.assertEqual(v.getvalue(), 0)
 
+    def test_document_insert_and_read(self):
+        col = self.soda.createCollection("it_docs")
+        # insertOneAndGet returns server-assigned key / version / media type
+        saved = col.insertOneAndGet(self.soda.createDocument({"name": "alice",
+                                                              "age": 30}))
+        self.assertTrue(saved.key)
+        self.assertTrue(saved.version)
+        self.assertEqual(saved.mediaType, "application/json")
+        self.assertIsNone(saved.getContentAsBytes())     # no content returned
+
+        # read it back by key -> content parsed as JSON
+        got = col.find().key(saved.key).getOne()
+        self.assertEqual(got.key, saved.key)
+        content = got.getContent()
+        self.assertEqual(content["name"], "alice")
+        self.assertEqual(content["age"], 30)
+        self.assertIsInstance(got.getContentAsString(), str)
+
+        # insertOne accepts a bare value; a valid-but-absent key reads as None
+        col.insertOne({"name": "bob"})
+        self.assertIsNone(col.find().key("AABBCCDDEEFF00112233445566").getOne())
+
+    def test_large_document_insert_and_read_guard(self):
+        col = self.soda.createCollection("it_big")
+        big = {"blob": "x" * 40000}              # > 32767-byte inline read limit
+        saved = col.insertOneAndGet(big)         # insert is fine (temp LOB)
+        self.assertTrue(saved.key)
+        with self.assertRaises(oracle.NotSupportedError):
+            col.find().key(saved.key).getOne()   # read guards against truncation
+
 
 @unittest.skipUnless(_USER, _SKIP_REASON)
 class SqlDomainIntegration(_IntegrationBase):
@@ -3150,6 +3180,13 @@ async def test_async_soda_collections(self):
             self.assertIn("contentColumn", await col.get_metadata())
             self.assertEqual(await soda.getCollectionNames(), ["it_async"])
             self.assertIsNone(await soda.openCollection("no_such_coll"))
+            # document insert + read by key on the async path (#200)
+            saved = await col.insertOneAndGet(soda.createDocument({"x": 1}))
+            self.assertTrue(saved.key)
+            got = await col.find().key(saved.key).getOne()
+            self.assertEqual(got.getContent()["x"], 1)
+            self.assertIsNone(
+                await col.find().key("AABBCCDDEEFF00112233445566").getOne())
             self.assertTrue(await col.drop())
             self.assertFalse(await col.drop())