Add set_last_insert_rowid

doc/changes.rst

@@ -10,6 +10,8 @@ Updated completions in shell (eg added pragmas).
 `Resumable Bulk Update (RBU) <https://www.sqlite.org/rbu.html>`__
 extension is now built by default for :doc:`--enable-all-extensions <build>`.
 
+Added :meth:`Connection.set_last_insert_rowid`.
+
 3.17.0-r1
 =========
 

src/connection.c

@@ -806,6 +806,32 @@ Connection_last_insert_rowid(Connection *self)
   return PyLong_FromLongLong(sqlite3_last_insert_rowid(self->db));
 }
 
+/** .. method:: set_last_insert_rowid(int)
+
+  Sets the value calls to :meth:`last_insert_rowid` will return.
+
+  -* sqlite3_set_last_insert_rowid
+*/
+static PyObject *
+Connection_set_last_insert_rowid(Connection *self, PyObject *o)
+{
+  sqlite3_int64 rowid;
+
+  CHECK_USE(NULL);
+  CHECK_CLOSED(self,NULL);
+
+  if (!PyIntLong_Check(o))
+    return PyErr_Format(PyExc_TypeError, "rowid should be 64bit number");
+
+  rowid=PyIntLong_AsLongLong(o);
+  if(PyErr_Occurred())
+    return NULL;
+
+  PYSQLITE_VOID_CALL(sqlite3_set_last_insert_rowid(self->db, rowid));
+
+  Py_RETURN_NONE;
+}
+
 
 /** .. method:: interrupt()
 
@@ -3353,6 +3379,8 @@ static PyMethodDef Connection_methods[] = {
    "Creates a collation function"},
   {"last_insert_rowid", (PyCFunction)Connection_last_insert_rowid, METH_NOARGS,
    "Returns rowid for last insert"},
+  {"set_last_insert_rowid", (PyCFunction)Connection_set_last_insert_rowid, METH_O,
+   "Sets rowid returned for for last insert_rowid"},
   {"collationneeded", (PyCFunction)Connection_collationneeded, METH_O,
    "Sets collation needed callback"},
   {"setauthorizer", (PyCFunction)Connection_setauthorizer, METH_O,

src/pyutil.c

@@ -88,11 +88,13 @@ typedef int Py_ssize_t;
 #define PyBytesObject             PyStringObject
 #define PyIntLong_Check(x)        (PyInt_Check((x)) || PyLong_Check((x)))
 #define PyIntLong_AsLong(x)       ( (PyInt_Check((x))) ? ( PyInt_AsLong((x)) ) : ( (PyLong_AsLong((x)))))
+#define PyIntLong_AsLongLong(x)   ( (PyInt_Check((x))) ? ( PyInt_AsLong((x)) ) : ( (PyLong_AsLongLong((x)))))
 #define PyBytes_FromFormat        PyString_FromFormat
 #else
 #define PyIntLong_Check           PyLong_Check
 #define PyIntLong_AsLong          PyLong_AsLong
 #define PyInt_FromLong            PyLong_FromLong
+#define PyIntLong_AsLongLong      PyLong_AsLongLong
 #define PyObject_Unicode          PyObject_Str
 #endif
 

tests.py

@@ -314,7 +314,8 @@ class APSW(unittest.TestCase):
         'wal_autocheckpoint': 1,
         'setwalhook': 1,
         'readonly': 1,
-        'db_filename': 1
+        'db_filename': 1,
+        'set_last_insert_rowid': 1,
         }
 
     cursor_nargs={
@@ -1503,6 +1504,16 @@ def testLastInsertRowId(self):
         v=2**40
         c.execute("insert into foo values(?)", (v,))
         self.assertEqual(v, self.db.last_insert_rowid())
+        # try setting it
+        self.assertRaises(TypeError, self.db.set_last_insert_rowid,)
+        self.assertRaises(TypeError, self.db.set_last_insert_rowid, "3")
+        self.assertRaises(TypeError, self.db.set_last_insert_rowid, "3", 3)
+        self.assertRaises(OverflowError, self.db.set_last_insert_rowid, 2**40*2**40)
+        for v in -20, 0, 20, 2**32-1, -2**32-1, 2**60, -2**60:
+            c.execute("insert into foo values(?)", (v-3,))
+            self.assertNotEqual(v, self.db.last_insert_rowid())
+            self.db.set_last_insert_rowid(v)
+            self.assertEqual(v, self.db.last_insert_rowid())
 
     def testComplete(self):
         "Completeness of SQL statement checking"