Issue #13773: sqlite3.connect() gets a new `uri` parameter to pass the...

Issue #13773: sqlite3.connect() gets a new `uri` parameter to pass the filename as a URI, allowing to pass custom options.

Doc/library/sqlite3.rst

@@ -159,7 +159,7 @@ Module functions and constants
    first blank for the column name: the column name would simply be "x".
 
 
-.. function:: connect(database[, timeout, detect_types, isolation_level, check_same_thread, factory, cached_statements])
+.. function:: connect(database[, timeout, detect_types, isolation_level, check_same_thread, factory, cached_statements, uri])
 
    Opens a connection to the SQLite database file *database*. You can use
    ``":memory:"`` to open a database connection to a database that resides in RAM
@@ -195,6 +195,18 @@ Module functions and constants
    for the connection, you can set the *cached_statements* parameter. The currently
    implemented default is to cache 100 statements.
 
+   If *uri* is true, *database* is interpreted as a URI. This allows you
+   to specify options. For example, to open a database in read-only mode
+   you can use::
+
+       db = sqlite3.connect('file:path/to/database?mode=ro', uri=True)
+
+   More information about this feature, including a list of recognized options, can
+   be found in the `SQLite URI documentation <http://www.sqlite.org/uri.html>`_.
+
+   .. versionchanged:: 3.4
+      Added the *uri* parameter.
+
 
 .. function:: register_converter(typename, callable)
 

Lib/sqlite3/test/dbapi.py

@@ -28,6 +28,9 @@ try:
 except ImportError:
     threading = None
 
+from test.support import TESTFN, unlink
+
+
 class ModuleTests(unittest.TestCase):
     def CheckAPILevel(self):
         self.assertEqual(sqlite.apilevel, "2.0",
@@ -163,6 +166,21 @@ class ConnectionTests(unittest.TestCase):
         with self.assertRaises(AttributeError):
             self.cx.in_transaction = True
 
+    def CheckOpenUri(self):
+        if sqlite.sqlite_version_info < (3, 7, 7):
+            with self.assertRaises(sqlite.NotSupportedError):
+                sqlite.connect(':memory:', uri=True)
+            return
+        self.addCleanup(unlink, TESTFN)
+        with sqlite.connect(TESTFN) as cx:
+            cx.execute('create table test(id integer)')
+        with sqlite.connect('file:' + TESTFN, uri=True) as cx:
+            cx.execute('insert into test(id) values(0)')
+        with sqlite.connect('file:' + TESTFN + '?mode=ro', uri=True) as cx:
+            with self.assertRaises(sqlite.OperationalError):
+                cx.execute('insert into test(id) values(1)')
+
+
 class CursorTests(unittest.TestCase):
     def setUp(self):
         self.cx = sqlite.connect(":memory:")

Misc/NEWS

@@ -244,6 +244,9 @@ Core and Builtins
 Library
 -------
 
+- Issue #13773: sqlite3.connect() gets a new `uri` parameter to pass the
+  filename as a URI, allowing to pass custom options.
+
 - Issue #17156: pygettext.py now uses an encoding of source file and correctly
   writes and escapes non-ascii characters.
 

Modules/_sqlite/connection.c

@@ -60,7 +60,11 @@ static void _sqlite3_result_error(sqlite3_context* ctx, const char* errmsg, int
 
 int pysqlite_connection_init(pysqlite_Connection* self, PyObject* args, PyObject* kwargs)
 {
-    static char *kwlist[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", NULL, NULL};
+    static char *kwlist[] = {
+        "database", "timeout", "detect_types", "isolation_level",
+        "check_same_thread", "factory", "cached_statements", "uri",
+        NULL
+    };
 
     char* database;
     int detect_types = 0;
@@ -68,11 +72,14 @@ int pysqlite_connection_init(pysqlite_Connection* self, PyObject* args, PyObject
     PyObject* factory = NULL;
     int check_same_thread = 1;
     int cached_statements = 100;
+    int uri = 0;
     double timeout = 5.0;
     int rc;
 
-    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOi", kwlist,
-                                     &database, &timeout, &detect_types, &isolation_level, &check_same_thread, &factory, &cached_statements))
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOip", kwlist,
+                                     &database, &timeout, &detect_types,
+                                     &isolation_level, &check_same_thread,
+                                     &factory, &cached_statements, &uri))
     {
         return -1;
     }
@@ -91,8 +98,19 @@ int pysqlite_connection_init(pysqlite_Connection* self, PyObject* args, PyObject
     Py_INCREF(&PyUnicode_Type);
     self->text_factory = (PyObject*)&PyUnicode_Type;
 
+#ifdef SQLITE_OPEN_URI
+    Py_BEGIN_ALLOW_THREADS
+    rc = sqlite3_open_v2(database, &self->db,
+                         SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE |
+                         (uri ? SQLITE_OPEN_URI : 0), NULL);
+#else
+    if (uri) {
+        PyErr_SetString(pysqlite_NotSupportedError, "URIs not supported");
+        return -1;
+    }
     Py_BEGIN_ALLOW_THREADS
     rc = sqlite3_open(database, &self->db);
+#endif
     Py_END_ALLOW_THREADS
 
     if (rc != SQLITE_OK) {

Modules/_sqlite/module.c

@@ -50,19 +50,26 @@ static PyObject* module_connect(PyObject* self, PyObject* args, PyObject*
      * C-level, so this code is redundant with the one in connection_init in
      * connection.c and must always be copied from there ... */
 
-    static char *kwlist[] = {"database", "timeout", "detect_types", "isolation_level", "check_same_thread", "factory", "cached_statements", NULL, NULL};
+    static char *kwlist[] = {
+        "database", "timeout", "detect_types", "isolation_level",
+        "check_same_thread", "factory", "cached_statements", "uri",
+        NULL
+    };
     char* database;
     int detect_types = 0;
     PyObject* isolation_level;
     PyObject* factory = NULL;
     int check_same_thread = 1;
     int cached_statements;
+    int uri = 0;
     double timeout = 5.0;
 
     PyObject* result;
 
-    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOi", kwlist,
-                                     &database, &timeout, &detect_types, &isolation_level, &check_same_thread, &factory, &cached_statements))
+    if (!PyArg_ParseTupleAndKeywords(args, kwargs, "s|diOiOip", kwlist,
+                                     &database, &timeout, &detect_types,
+                                     &isolation_level, &check_same_thread,
+                                     &factory, &cached_statements, &uri))
     {
         return NULL;
     }
@@ -77,7 +84,8 @@ static PyObject* module_connect(PyObject* self, PyObject* args, PyObject*
 }
 
 PyDoc_STRVAR(module_connect_doc,
-"connect(database[, timeout, isolation_level, detect_types, factory])\n\
+"connect(database[, timeout, detect_types, isolation_level,\n\
+        check_same_thread, factory, cached_statements, uri])\n\
 \n\
 Opens a connection to the SQLite database file *database*. You can use\n\
 \":memory:\" to open a database connection to a database that resides in\n\