gh-69093: Support basic incremental I/O to blobs in sqlite3 (GH-30680)

Authored-by: Aviv Palivoda <[email protected]>
Co-authored-by: Erlend E. Aasland <[email protected]>
Co-authored-by: palaviv <[email protected]>
Co-authored-by: Kumar Aditya <[email protected]>
Co-authored-by: Jelle Zijlstra <[email protected]>

Author: 4 people

Doc/includes/sqlite3/blob.py

@@ -0,0 +1,12 @@
+import sqlite3
+
+con = sqlite3.connect(":memory:")
+con.execute("create table test(blob_col blob)")
+con.execute("insert into test(blob_col) values (zeroblob(10))")
+
+blob = con.blobopen("test", "blob_col", 1)
+blob.write(b"Hello")
+blob.write(b"World")
+blob.seek(0)
+print(blob.read())  # will print b"HelloWorld"
+blob.close()

Doc/library/sqlite3.rst

@@ -394,6 +394,20 @@ Connection Objects
       supplied, this must be a callable returning an instance of :class:`Cursor`
       or its subclasses.
 
+   .. method:: blobopen(table, column, row, /, *, readonly=False, name="main")
+
+      Open a :class:`Blob` handle to the :abbr:`BLOB (Binary Large OBject)`
+      located in row *row*, column *column*, table *table* of database *name*.
+      When *readonly* is :const:`True` the blob is opened without write
+      permissions.
+
+      .. note::
+
+         The blob size cannot be changed using the :class:`Blob` class.
+         Use the SQL function ``zeroblob`` to create a blob with a fixed size.
+
+      .. versionadded:: 3.11
+
    .. method:: commit()
 
       This method commits the current transaction. If you don't call this method,
@@ -1088,6 +1102,58 @@ Exceptions
    transactions turned off.  It is a subclass of :exc:`DatabaseError`.
 
 
+.. _sqlite3-blob-objects:
+
+Blob Objects
+------------
+
+.. versionadded:: 3.11
+
+.. class:: Blob
+
+   A :class:`Blob` instance is a :term:`file-like object` that can read and write
+   data in an SQLite :abbr:`BLOB (Binary Large OBject)`.  Call ``len(blob)`` to
+   get the size (number of bytes) of the blob.
+
+   .. method:: close()
+
+      Close the blob.
+
+      The blob will be unusable from this point onward.  An
+      :class:`~sqlite3.Error` (or subclass) exception will be raised if any
+      further operation is attempted with the blob.
+
+   .. method:: read(length=-1, /)
+
+      Read *length* bytes of data from the blob at the current offset position.
+      If the end of the blob is reached, the data up to
+      :abbr:`EOF (End of File)` will be returned.  When *length* is not
+      specified, or is negative, :meth:`~Blob.read` will read until the end of
+      the blob.
+
+   .. method:: write(data, /)
+
+      Write *data* to the blob at the current offset.  This function cannot
+      change the blob length.  Writing beyond the end of the blob will raise
+      :exc:`ValueError`.
+
+   .. method:: tell()
+
+      Return the current access position of the blob.
+
+   .. method:: seek(offset, origin=os.SEEK_SET, /)
+
+      Set the current access position of the blob to *offset*.  The *origin*
+      argument defaults to :data:`os.SEEK_SET` (absolute blob positioning).
+      Other values for *origin* are :data:`os.SEEK_CUR` (seek relative to the
+      current position) and :data:`os.SEEK_END` (seek relative to the blob’s
+      end).
+
+   :class:`Blob` example:
+
+      .. literalinclude:: ../includes/sqlite3/blob.py
+
+
 .. _sqlite3-types:
 
 SQLite and Python types

Doc/whatsnew/3.11.rst

@@ -393,6 +393,10 @@ sqlite3
   :class:`sqlite3.Connection` for creating aggregate window functions.
   (Contributed by Erlend E. Aasland in :issue:`34916`.)
 
+* Add :meth:`~sqlite3.Connection.blobopen` to :class:`sqlite3.Connection`.
+  :class:`sqlite3.Blob` allows incremental I/O operations on blobs.
+  (Contributed by Aviv Palivoda and Erlend E. Aasland in :issue:`24905`)
+
 
 sys
 ---

Lib/test/test_sqlite3/test_dbapi.py

@@ -33,6 +33,8 @@
     check_disallow_instantiation,
     threading_helper,
 )
+from _testcapi import INT_MAX
+from os import SEEK_SET, SEEK_CUR, SEEK_END
 from test.support.os_helper import TESTFN, unlink, temp_dir
 
 
@@ -1041,11 +1043,163 @@ def test_same_query_in_multiple_cursors(self):
             self.assertEqual(cu.fetchall(), [(1,)])
 
 
+class BlobTests(unittest.TestCase):
+    def setUp(self):
+        self.cx = sqlite.connect(":memory:")
+        self.cx.execute("create table test(b blob)")
+        self.data = b"this blob data string is exactly fifty bytes long!"
+        self.cx.execute("insert into test(b) values (?)", (self.data,))
+        self.blob = self.cx.blobopen("test", "b", 1)
+
+    def tearDown(self):
+        self.blob.close()
+        self.cx.close()
+
+    def test_blob_seek_and_tell(self):
+        self.blob.seek(10)
+        self.assertEqual(self.blob.tell(), 10)
+
+        self.blob.seek(10, SEEK_SET)
+        self.assertEqual(self.blob.tell(), 10)
+
+        self.blob.seek(10, SEEK_CUR)
+        self.assertEqual(self.blob.tell(), 20)
+
+        self.blob.seek(-10, SEEK_END)
+        self.assertEqual(self.blob.tell(), 40)
+
+    def test_blob_seek_error(self):
+        msg_oor = "offset out of blob range"
+        msg_orig = "'origin' should be os.SEEK_SET, os.SEEK_CUR, or os.SEEK_END"
+        msg_of = "seek offset results in overflow"
+
+        dataset = (
+            (ValueError, msg_oor, lambda: self.blob.seek(1000)),
+            (ValueError, msg_oor, lambda: self.blob.seek(-10)),
+            (ValueError, msg_orig, lambda: self.blob.seek(10, -1)),
+            (ValueError, msg_orig, lambda: self.blob.seek(10, 3)),
+        )
+        for exc, msg, fn in dataset:
+            with self.subTest(exc=exc, msg=msg, fn=fn):
+                self.assertRaisesRegex(exc, msg, fn)
+
+        # Force overflow errors
+        self.blob.seek(1, SEEK_SET)
+        with self.assertRaisesRegex(OverflowError, msg_of):
+            self.blob.seek(INT_MAX, SEEK_CUR)
+        with self.assertRaisesRegex(OverflowError, msg_of):
+            self.blob.seek(INT_MAX, SEEK_END)
+
+    def test_blob_read(self):
+        buf = self.blob.read()
+        self.assertEqual(buf, self.data)
+
+    def test_blob_read_oversized(self):
+        buf = self.blob.read(len(self.data) * 2)
+        self.assertEqual(buf, self.data)
+
+    def test_blob_read_advance_offset(self):
+        n = 10
+        buf = self.blob.read(n)
+        self.assertEqual(buf, self.data[:n])
+        self.assertEqual(self.blob.tell(), n)
+
+    def test_blob_read_at_offset(self):
+        self.blob.seek(10)
+        self.assertEqual(self.blob.read(10), self.data[10:20])
+
+    def test_blob_read_error_row_changed(self):
+        self.cx.execute("update test set b='aaaa' where rowid=1")
+        with self.assertRaises(sqlite.OperationalError):
+            self.blob.read()
+
+    def test_blob_write(self):
+        new_data = b"new data".ljust(50)
+        self.blob.write(new_data)
+        row = self.cx.execute("select b from test").fetchone()
+        self.assertEqual(row[0], new_data)
+
+    def test_blob_write_at_offset(self):
+        new_data = b"c" * 25
+        self.blob.seek(25)
+        self.blob.write(new_data)
+        row = self.cx.execute("select b from test").fetchone()
+        self.assertEqual(row[0], self.data[:25] + new_data)
+
+    def test_blob_write_advance_offset(self):
+        self.blob.write(b"d"*10)
+        self.assertEqual(self.blob.tell(), 10)
+
+    def test_blob_write_error_length(self):
+        with self.assertRaisesRegex(ValueError, "data longer than blob"):
+            self.blob.write(b"a" * 1000)
+
+    def test_blob_write_error_row_changed(self):
+        self.cx.execute("update test set b='aaaa' where rowid=1")
+        with self.assertRaises(sqlite.OperationalError):
+            self.blob.write(b"aaa")
+
+    def test_blob_write_error_readonly(self):
+        ro_blob = self.cx.blobopen("test", "b", 1, readonly=True)
+        with self.assertRaisesRegex(sqlite.OperationalError, "readonly"):
+            ro_blob.write(b"aaa")
+        ro_blob.close()
+
+    def test_blob_open_error(self):
+        dataset = (
+            (("test", "b", 1), {"name": "notexisting"}),
+            (("notexisting", "b", 1), {}),
+            (("test", "notexisting", 1), {}),
+            (("test", "b", 2), {}),
+        )
+        regex = "no such"
+        for args, kwds in dataset:
+            with self.subTest(args=args, kwds=kwds):
+                with self.assertRaisesRegex(sqlite.OperationalError, regex):
+                    self.cx.blobopen(*args, **kwds)
+
+    def test_blob_sequence_not_supported(self):
+        with self.assertRaises(TypeError):
+            self.blob + self.blob
+        with self.assertRaises(TypeError):
+            self.blob * 5
+        with self.assertRaises(TypeError):
+            b"a" in self.blob
+
+    def test_blob_closed(self):
+        with memory_database() as cx:
+            cx.execute("create table test(b blob)")
+            cx.execute("insert into test values (zeroblob(100))")
+            blob = cx.blobopen("test", "b", 1)
+            blob.close()
+
+            msg = "Cannot operate on a closed blob"
+            with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+                blob.read()
+            with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+                blob.write(b"")
+            with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+                blob.seek(0)
+            with self.assertRaisesRegex(sqlite.ProgrammingError, msg):
+                blob.tell()
+
+    def test_blob_closed_db_read(self):
+        with memory_database() as cx:
+            cx.execute("create table test(b blob)")
+            cx.execute("insert into test(b) values (zeroblob(100))")
+            blob = cx.blobopen("test", "b", 1)
+            cx.close()
+            self.assertRaisesRegex(sqlite.ProgrammingError,
+                                   "Cannot operate on a closed database",
+                                   blob.read)
+
+
 class ThreadTests(unittest.TestCase):
     def setUp(self):
         self.con = sqlite.connect(":memory:")
         self.cur = self.con.cursor()
-        self.cur.execute("create table test(name text)")
+        self.cur.execute("create table test(name text, b blob)")
+        self.cur.execute("insert into test values('blob', zeroblob(1))")
 
     def tearDown(self):
         self.cur.close()
@@ -1080,6 +1234,7 @@ def test_check_connection_thread(self):
             lambda: self.con.create_collation("foo", None),
             lambda: self.con.setlimit(sqlite.SQLITE_LIMIT_LENGTH, -1),
             lambda: self.con.getlimit(sqlite.SQLITE_LIMIT_LENGTH),
+            lambda: self.con.blobopen("test", "b", 1),
         ]
         if hasattr(sqlite.Connection, "serialize"):
             fns.append(lambda: self.con.serialize())

Misc/NEWS.d/next/Library/2018-04-18-16-15-55.bpo-24905.jYqjYx.rst

@@ -0,0 +1,3 @@
+Add :meth:`~sqlite3.Connection.blobopen` to :class:`sqlite3.Connection`.
+:class:`sqlite3.Blob` allows incremental I/O operations on blobs.
+Patch by Aviv Palivoda and Erlend E. Aasland.