Add Connection.db_names

Fixes #343

apsw/__init__.pyi

@@ -112,6 +112,7 @@ class Connection:
     def createscalarfunction(self, name: str, callable: Optional[ScalarProtocol], numargs: int = -1, deterministic: bool = False) -> None: ...
     def cursor(self) -> Cursor: ...
     def db_filename(self, name: str) -> str: ...
+    def db_names(self) -> List[str]: ...
     def deserialize(self, name: str, contents: bytes) -> None: ...
     def enableloadextension(self, enable: bool) -> None: ...
     def __enter__(self) -> Connection: ...

doc/changes.rst

@@ -21,6 +21,8 @@ Updated typing information with more detail (:issue:`338`)
 Corrected the :ref:`tips <diagnostics_tips>` log handler of extended
 result code (:issue:`342`)
 
+Added :func:`Connection.db_names` (:issue:`343`)
+
 3.38.5-r1
 =========
 

src/apsw.docstrings

@@ -840,6 +840,13 @@
 } while(0)
 
 
+#define  Connection_db_names_DOC "db_names($self)\n--\n\nConnection.db_names() -> List[str]\n\n" \
+"Returns the list of database names.  For example the first database\n" \
+"is named 'main', the next 'temp', and the rest with the name provided\n" \
+"in `ATTACH <https://www.sqlite.org/lang_attach.html>`__\n" \
+"\n" \
+"Calls: `sqlite3_db_name <https://sqlite.org/c3ref/db_name.html>`__\n" 
+
 #define  Connection_deserialize_DOC "deserialize($self,name,contents)\n--\n\nConnection.deserialize(name: str, contents: bytes) -> None\n\n" \
 "Replaces the named database with an in-memory copy of *contents*.\n" \
 "*name* is **\"main\"** for the main database, **\"temp\"** for the\n" \

src/connection.c

@@ -800,6 +800,56 @@ Connection_getautocommit(Connection *self)
   Py_RETURN_FALSE;
 }
 
+/** .. method:: db_names() -> List[str]
+
+ Returns the list of database names.  For example the first database
+ is named 'main', the next 'temp', and the rest with the name provided
+ in `ATTACH <https://www.sqlite.org/lang_attach.html>`__
+
+ -* sqlite3_db_name
+*/
+static PyObject *
+Connection_db_names(Connection *self)
+{
+  PyObject *res = NULL, *str = NULL;
+  int i;
+
+  CHECK_USE(NULL);
+  CHECK_CLOSED(self, NULL);
+
+  sqlite3_mutex_enter(sqlite3_db_mutex(self->db));
+
+  APSW_FAULT_INJECT(dbnamesnolist, res = PyList_New(0), res = PyErr_NoMemory());
+  if (!res)
+    goto error;
+
+  for (i = 0; i < APSW_INT32_MAX; i++)
+  {
+    int appendres;
+
+    const char *s = sqlite3_db_name(self->db, i); /* Doesn't need PYSQLITE_CALL */
+    if (!s)
+      break;
+    APSW_FAULT_INJECT(dbnamestrfail, str = convertutf8string(s), str = PyErr_NoMemory());
+    if (!str)
+      goto error;
+    APSW_FAULT_INJECT(dbnamesappendfail, appendres = PyList_Append(res, str), (PyErr_NoMemory(), appendres = -1));
+    if (0 != appendres)
+      goto error;
+    Py_CLEAR(str);
+  }
+
+  sqlite3_mutex_leave(sqlite3_db_mutex(self->db));
+  return res;
+error:
+  sqlite3_mutex_leave(sqlite3_db_mutex(self->db));
+  assert(PyErr_Occurred());
+  Py_XDECREF(res);
+  Py_XDECREF(str);
+
+  return NULL;
+}
+
 /** .. method:: last_insert_rowid() -> int
 
   Returns the integer key of the most recent insert in the database.
@@ -3663,6 +3713,8 @@ static PyMethodDef Connection_methods[] = {
      Connection_deserialize_DOC},
     {"autovacuum_pages", (PyCFunction)Connection_autovacuum_pages, METH_VARARGS | METH_KEYWORDS,
      Connection_autovacuum_pages_DOC},
+    {"db_names", (PyCFunction)Connection_db_names, METH_NOARGS,
+     Connection_db_names_DOC},
     {0, 0, 0, 0} /* Sentinel */
 };
 

tests.py

@@ -444,6 +444,31 @@ def testConnectionConfig(self):
             self.assertEqual(1, self.db.config(i, -1))
             self.assertEqual(0, self.db.config(i, 0))
 
+    def testConnectionNames(self):
+        "Test Connection.db_names"
+        self.assertRaises(TypeError, self.db.db_names, 3)
+        expected = ["main", "temp"]
+        self.assertEqual(expected, self.db.db_names())
+        for t in "", APSW.wikipedia_text:
+            self.db.cursor().execute(f"attach '{ self.db.db_filename('main') }' as '{ t }'")
+            expected.append(t)
+        self.assertEqual(expected, self.db.db_names())
+        while True:
+            t = f"{ expected[-1] }-{ len(expected) }"
+            try:
+                self.db.cursor().execute(f"attach '{ self.db.db_filename('main') }' as '{ t }'")
+            except apsw.SQLError:
+                # SQLError: too many attached databases - max ....
+                break
+            expected.append(t)
+        self.assertEqual(expected, self.db.db_names())
+        while len(expected)>2:
+            i =random.randint(2, len(expected)-1)
+            self.db.cursor().execute(f"detach '{ expected[i] }'")
+            del expected[i]
+            self.assertEqual(expected, self.db.db_names())
+
+
     def testMemoryLeaks(self):
         "MemoryLeaks: Run with a memory profiler such as valgrind and debug Python"
         # make and toss away a bunch of db objects, cursors, functions etc - if you use memory profiling then
@@ -3754,7 +3779,7 @@ def testWikipedia(self):
            # is already held by enclosing sqlite3_step and the
            # methods will only be called from that same thread so it
            # isn't a problem.
-                        'skipcalls': re.compile("^sqlite3_(blob_bytes|column_count|bind_parameter_count|data_count|vfs_.+|changes64|total_changes64|get_autocommit|last_insert_rowid|complete|interrupt|limit|malloc64|free|threadsafe|value_.+|libversion|enable_shared_cache|initialize|shutdown|config|memory_.+|soft_heap_limit(64)?|randomness|db_readonly|db_filename|release_memory|status64|result_.+|user_data|mprintf|aggregate_context|declare_vtab|backup_remaining|backup_pagecount|sourceid|uri_.+)$"),
+                        'skipcalls': re.compile("^sqlite3_(blob_bytes|column_count|bind_parameter_count|data_count|vfs_.+|changes64|total_changes64|get_autocommit|last_insert_rowid|complete|interrupt|limit|malloc64|free|threadsafe|value_.+|libversion|enable_shared_cache|initialize|shutdown|config|memory_.+|soft_heap_limit(64)?|randomness|db_readonly|db_filename|release_memory|status64|result_.+|user_data|mprintf|aggregate_context|declare_vtab|backup_remaining|backup_pagecount|mutex_enter|mutex_leave|sourceid|uri_.+)$"),
                         # error message
                         'desc': "sqlite3_ calls must wrap with PYSQLITE_CALL",
                         },
@@ -8362,6 +8387,15 @@ def foo():
         except apsw.FullError:
             pass
 
+        # Connection.db_names
+        apsw.faultdict["dbnamesnolist"] = True
+        self.assertRaises(MemoryError, self.db.db_names)
+        apsw.faultdict["dbnamestrfail"] = True
+        self.assertRaises(MemoryError, self.db.db_names)
+        apsw.faultdict["dbnamesappendfail"] = True
+        self.assertRaises(MemoryError, self.db.db_names)
+
+
     # This test is run last by deliberate name choice.  If it did
     # uncover any bugs there isn't much that can be done to turn the
     # checker off.