python · MengAiDev · Aug 20, 2025 · Aug 20, 2025 · Aug 20, 2025 · ssmycelium
@@ -1510,6 +1510,55 @@ Cursor objects
 
       Use :meth:`executescript` to execute multiple SQL statements.
 
+   .. method:: execute_json(sql, parameters=(), /)
+
+      Execute a single SQL statement and return the result as a JSON string.
+      This method is similar to :meth:`execute` but returns the query results
+      directly as a JSON string instead of rows.
+
+      :param str sql:
+         A single SQL statement that returns rows.
+
+      :param parameters:
+         Python values to bind to placeholders in *sql*.
+         A :class:`!dict` if named placeholders are used.
+         A :term:`!sequence` if unnamed placeholders are used.
+         See :ref:`sqlite3-placeholders`.
+      :type parameters: :class:`dict` | :term:`sequence`
+
+      :raises ProgrammingError:
+         When *sql* contains more than one SQL statement.
+         When :ref:`named placeholders <sqlite3-placeholders>` are used
+         and *parameters* is a sequence instead of a :class:`dict`.
+
+      This method provides significant performance benefits when working with
+      JSON data, as it leverages SQLite's built-in JSON functions to generate
+      the JSON directly in the database engine rather than fetching rows and
+      converting them to JSON in Python.
+
+      Example:
+
+      .. testcode:: sqlite3.cursor
+
+         # Create and populate a table
+         cur.execute("CREATE TABLE users(id INTEGER, name TEXT, email TEXT)")
+         cur.executemany("INSERT INTO users VALUES(?, ?, ?)", [
+             (1, 'Alice', 'alice@example.com'),
+             (2, 'Bob', 'bob@example.com')
+         ])
+
+         # Get results as JSON
+         json_result = cur.execute_json("SELECT * FROM users")
+         print(json_result)
+         # Output: [{"id":1,"name":"Alice","email":"alice@example.com"},{"id":2,"name":"Bob","email":"bob@example.com"}]
+
+      .. testoutput:: sqlite3.cursor
+         :hide:
+
+         [{"id":1,"name":"Alice","email":"alice@example.com"},{"id":2,"name":"Bob","email":"bob@example.com"}]
+
+      .. versionadded:: 3.15
+
    .. method:: executemany(sql, parameters, /)
 
       For every item in *parameters*,

@@ -0,0 +1,103 @@
+# pysqlite2/test/execute_json.py: tests for execute_json method
+#
+# Copyright (C) 2025 Python Software Foundation
+#
+# This file is part of pysqlite.
+#
+# This software is provided 'as-is', without any express or implied
+# warranty.  In no event will the authors be held liable for any damages
+# arising from the use of this software.
+#
+# Permission is granted to anyone to use this software for any purpose,
+# including commercial applications, and to alter it and redistribute it
+# freely, subject to the following restrictions:
+#
+# 1. The origin of this software must not be misrepresented; you must not
+#    claim that you wrote the original software. If you use this software
+#    in a product, an acknowledgment in the product documentation would be
+#    appreciated but is not required.
+# 2. Altered source versions must be plainly marked as such, and must not be
+#    misrepresented as being the original software.
+# 3. This notice may not be removed or altered from any source distribution.
+
+import json
+import sqlite3
+import unittest
+
+
+
+class ExecuteJsonTests(unittest.TestCase):
+    def setUp(self):
+        self.cx = sqlite3.connect(":memory:")
+        self.cu = self.cx.cursor()
+        self.cu.execute(
+            "create table test(id integer primary key, name text, "
+            "income number, unique_test text unique)"
+        )
+        self.cu.execute("insert into test(name, income) values (?, ?)", ("foo", 100))
+        self.cu.execute("insert into test(name, income) values (?, ?)", ("bar", 200))
+        self.cu.execute("insert into test(name, income) values (?, ?)", ("baz", 300))
+
+    def tearDown(self):
+        self.cu.close()
+        self.cx.close()
+
+    def test_execute_json_basic(self):
+        # Test basic functionality of execute_json
+        result = self.cu.execute_json("select * from test")
+        data = json.loads(result)
+        self.assertIsInstance(data, list)
+        self.assertEqual(len(data), 3)
+        self.assertEqual(data[0]["name"], "foo")
+        self.assertEqual(data[0]["income"], 100)
+        self.assertEqual(data[1]["name"], "bar")
+        self.assertEqual(data[1]["income"], 200)
+        self.assertEqual(data[2]["name"], "baz")
+        self.assertEqual(data[2]["income"], 300)
+
+    def test_execute_json_empty_result(self):
+        # Test execute_json with empty result set
+        result = self.cu.execute_json("select * from test where id > 1000")
+        data = json.loads(result)
+        self.assertIsInstance(data, list)
+        self.assertEqual(len(data), 0)
+
+    def test_execute_json_with_parameters(self):
+        # Test execute_json with parameterized queries
+        result = self.cu.execute_json("select * from test where income > ?", (150,))
+        data = json.loads(result)
+        self.assertIsInstance(data, list)
+        self.assertEqual(len(data), 2)
+        self.assertEqual(data[0]["name"], "bar")
+        self.assertEqual(data[0]["income"], 200)
+        self.assertEqual(data[1]["name"], "baz")
+        self.assertEqual(data[1]["income"], 300)
+
+    def test_execute_json_with_named_parameters(self):
+        # Test execute_json with named parameters
+        result = self.cu.execute_json("select * from test where income > :min_income",
+                                     {"min_income": 150})
+        data = json.loads(result)
+        self.assertIsInstance(data, list)
+        self.assertEqual(len(data), 2)
+        self.assertEqual(data[0]["name"], "bar")
+        self.assertEqual(data[0]["income"], 200)
+        self.assertEqual(data[1]["name"], "baz")
+        self.assertEqual(data[1]["income"], 300)
+
+    def test_execute_json_invalid_sql(self):
+        # Test execute_json with invalid SQL
+        with self.assertRaises(sqlite3.OperationalError):
+            self.cu.execute_json("select asdf")
+
+    def test_execute_json_non_select(self):
+        # Test execute_json with non-SELECT statement
+        result = self.cu.execute_json("insert into test(name, income) values (?, ?)",
+                                     ("new_entry", 400))
+        data = json.loads(result)
+        self.assertIsInstance(data, list)
+        self.assertEqual(len(data), 0)
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/Misc/NEWS.d/next/Library/2025-08-20-17-18-50.gh-issue-137974.L9WxPy.rst b/Misc/NEWS.d/next/Library/2025-08-20-17-18-50.gh-issue-137974.L9WxPy.rst
@@ -0,0 +1 @@
+JSON (output) support for sqlite3
@@ -977,6 +977,146 @@ _pysqlite_query_execute(pysqlite_Cursor* self, int multiple, PyObject* operation
     return Py_NewRef((PyObject *)self);
 }
 
+/*[clinic input]
+_sqlite3.Cursor.execute_json as pysqlite_cursor_execute_json
+
+    sql: unicode
+    parameters: object(c_default = 'NULL') = ()
+    /
+
+Executes an SQL statement and returns the result as a JSON string.
+[clinic start generated code]*/
+
+static PyObject *
+pysqlite_cursor_execute_json_impl(pysqlite_Cursor *self, PyObject *sql,
+                                  PyObject *parameters)
+/*[clinic end generated code: output=192c296377bf8175 input=3c6c8e94ef0b53c7]*/
+{
+    // First execute the query normally to get column information
+    PyObject *result = _pysqlite_query_execute(self, 0, sql, parameters);
+    if (!result) {
+        return NULL;
+    }
+    Py_DECREF(result);
+
+    // Check if we have a statement with results
+    if (!self->statement || sqlite3_column_count(self->statement->st) == 0) {
+        // No results to convert to JSON, return empty JSON array
+        return PyUnicode_FromString("[]");
+    }
+
+    // Build a JSON query that wraps the original query
+    sqlite3_stmt *stmt = self->statement->st;
+    int numcols = sqlite3_column_count(stmt);
+
+    // Build the json_object parameters
+    PyObject *column_list = PyList_New(0);
+    if (!column_list) {
+        return NULL;
+    }
+
+    for (int i = 0; i < numcols; i++) {
+        const char *colname = sqlite3_column_name(stmt, i);
+        if (!colname) {
+            Py_DECREF(column_list);
+            return PyErr_NoMemory();
+        }
+
+        // Add column name as quoted string literal for json_object keys
+        PyObject *colname_obj = PyUnicode_FromFormat("'%s'", colname);
+        if (!colname_obj) {
+            Py_DECREF(column_list);
+            return NULL;
+        }
+
+        if (PyList_Append(column_list, colname_obj) < 0) {
+            Py_DECREF(colname_obj);
+            Py_DECREF(column_list);
+            return NULL;
+        }
+        Py_DECREF(colname_obj);
+
+        // Add column reference
+        PyObject *colref_obj = PyUnicode_FromFormat("row.%s", colname);
+        if (!colref_obj) {
+            Py_DECREF(column_list);
+            return NULL;
+        }
+
+        if (PyList_Append(column_list, colref_obj) < 0) {
+            Py_DECREF(colref_obj);
+            Py_DECREF(column_list);
+            return NULL;
+        }
+        Py_DECREF(colref_obj);
+    }
+
+    // Join the column list with commas
+    PyObject *comma = PyUnicode_FromString(",");
+    if (!comma) {
+        Py_DECREF(column_list);
+        return NULL;
+    }
+
+    PyObject *column_str = PyUnicode_Join(comma, column_list);
+    Py_DECREF(comma);
+    if (!column_str) {
+        Py_DECREF(column_list);
+        return NULL;
+    }
+
+    // Complete the JSON query
+    PyObject *full_query = PyUnicode_FromFormat("SELECT json_group_array(json_object(%S)) FROM (%U) AS row", column_str, sql);
+    Py_DECREF(column_str);
+    Py_DECREF(column_list);
+    if (!full_query) {
+        return NULL;
+    }
+
+    // Execute the JSON query
+    pysqlite_Statement *json_stmt = pysqlite_statement_create(self->connection, full_query);
+    Py_DECREF(full_query);
+    if (!json_stmt) {
+        return NULL;
+    }
+
+    // Bind parameters if needed
+    if (parameters != NULL && parameters != Py_None) {
+        bind_parameters(self->connection->state, json_stmt, parameters);
+        if (PyErr_Occurred()) {
+            Py_DECREF(json_stmt);
+            return NULL;
+        }
+    }
+
+    // Execute the statement
+    int rc = stmt_step(json_stmt->st);
+    if (rc != SQLITE_ROW) {
+        Py_DECREF(json_stmt);
+        if (rc == SQLITE_DONE) {
+            // No rows returned, return empty JSON array
+            return PyUnicode_FromString("[]");
+        } else {
+            // Error occurred
+            set_error_from_db(self->connection->state, self->connection->db);
+            return NULL;
+        }
+    }
+
+    // Get the JSON result
+    const char *json_result = (const char*)sqlite3_column_text(json_stmt->st, 0);
+    PyObject *result_str = NULL;
+    if (json_result) {
+        result_str = PyUnicode_FromString(json_result);
+    } else {
+        // NULL result, return empty JSON array
+        result_str = PyUnicode_FromString("[]");
+    }
+
+    Py_DECREF(json_stmt);
+    return result_str;
+}
+
 /*[clinic input]
 _sqlite3.Cursor.execute as pysqlite_cursor_execute
 
@@ -1303,6 +1443,7 @@ pysqlite_cursor_close_impl(pysqlite_Cursor *self)
 
 static PyMethodDef cursor_methods[] = {
     PYSQLITE_CURSOR_CLOSE_METHODDEF
+    PYSQLITE_CURSOR_EXECUTE_JSON_METHODDEF
     PYSQLITE_CURSOR_EXECUTEMANY_METHODDEF
     PYSQLITE_CURSOR_EXECUTESCRIPT_METHODDEF
     PYSQLITE_CURSOR_EXECUTE_METHODDEF