bpo-39170: Sqlite3 row_factory for attribute access: sqlite3.NamedRow

Doc/library/sqlite3.rst

@@ -494,14 +494,21 @@ Connection Objects
       .. literalinclude:: ../includes/sqlite3/row_factory.py
 
       If returning a tuple doesn't suffice and you want name-based access to
-      columns, you should consider setting :attr:`row_factory` to the
-      highly-optimized :class:`sqlite3.Row` type. :class:`Row` provides both
-      index-based and case-insensitive name-based access to columns with almost no
-      memory overhead. It will probably be better than your own custom
-      dictionary-based approach or even a db_row based solution.
+      columns, you should consider setting :attr:`row_factory` one of the
+      highly optimized row_factory types :class:`sqlite3.Row` or
+      :class:`sqlite3.NamedRow`.
+
+      :class:`Row` provides both index-based and case-insensitive name-based
+      access to columns with almost no memory overhead. It will probably be
+      better than your own custom dictionary-based approach or even a db_row
+      based solution.
 
       .. XXX what's a db_row-based solution?
 
+      :class:`NamedRow` provides both attribute and index by column name or number
+      to columns with the same low memory requirements of :class:`sqlite3.Row`.
+      As a compiled class, it will out perform a custom python row_factory solution.
+
 
    .. attribute:: text_factory
 
@@ -808,6 +815,72 @@ Now we plug :class:`Row` in::
    100.0
    35.14
 
+.. class:: NamedRow
+
+   A :class:`NamedRow` instance is an optimized :attr:`~Connection.row_factory`
+   for :class:`Connection` objects.
+   It tries to mimic a namedtuple in most of its features.
+
+   It supports mapping access by attribute, column name, and index.
+   It also supports iteration, slicing, equality testing, :func:`contains`,
+   and :func:`len`.
+
+Here is a simple example, with a Japanese column name::
+
+    >>> db_connection = sqlite3.connect(":memory:")
+    >>> cursor = db_connection.cursor()
+    >>> cursor.row_factory = sqlite3.NamedRow
+    >>> row = cursor.execute("SELECT 'seven' AS english, '七' as 日本").fetchone()
+    >>> type(row)
+    <class 'sqlite3.NamedRow'>
+    >>> len(row)
+    2
+    >>> '日本' in row
+    True
+    >>> row.english, row.日本
+    ('seven', '七')
+    >>> row[0], row['english']
+    ('seven', 'seven')
+    >>> tuple(row)
+    (('english', 'seven'), ('日本', '七'))
+    >>> dict(row)
+    {'english': 'seven', '日本': '七'}
+    >>> [x[0] for x in row]  # Get column names
+    ['english', '日本']
+    >>> [x[1] for x in row]  # Get column values
+    ['seven', '七']
+
+For columns names that are invalid attribute names, such as "count(*)"
+or python reserved words, create a valid attribute name using the SQL
+"AS" keyword.
+For example, assume we have an address table and want the number of rows
+where the city is Kahlotus.::
+
+    >>> cursor = db_connection.cursor()
+    >>> cursor.row_factory = sqlite3.NamedRow
+    >>> row = cursor.execute("SELECT city, count(*) FROM address WHERE city=?",
+    ...  ('Kahlotus')).fetchone()
+    >>> row.count
+    Traceback (most recent call last):
+      File "<stdin>", line 1, in <module>
+      AttributeError: 'sqlite3.NamedRow' object has no attribute 'count'
+    >>> tuple(row)
+    (('city', 'Kahlotus'), ('count(*)', 2))
+    >>> row = cursor.execute("SELECT city, count(*) AS cnt FROM address WHERE city=?",
+    ...   ('Kahlotus')).fetchone()
+    (('city', 'Kahlotus'), ('count', 2))
+    >>> row.count
+    2
+
+If you are unable to change the query, index by name or number.::
+
+    >>> tuple(row)
+    (('city', 'Kahlotus'), ('count(*)', 2))
+    >>> row['count(*)']
+    2
+    >>> row[1]
+    2
+
 
 .. _sqlite3-exceptions:
 

Lib/sqlite3/dbapi2.py

@@ -52,6 +52,8 @@ def TimestampFromTicks(ticks):
 
 Binary = memoryview
 collections.abc.Sequence.register(Row)
+collections.abc.Sequence.register(NamedRow)
+
 
 def register_adapters_and_converters():
     def adapt_date(val):

Lib/sqlite3/test/test_namedrow.py

@@ -0,0 +1,178 @@
+# -*- coding: utf-8 -*-
+# Tests the NamedRow row_factory with uft-8 for non ascii attribute names
+#
+# Copyright (C) 2019 Clinton James "JIDN"
+#
+# This software is provided 'as-is', without any express or implied
+# warranty.  In no even 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. Ths origin of this software must to 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 version 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 unittest
+import sqlite3
+
+
+class NamedRowFactoryTests(unittest.TestCase):
+    def setUp(self):
+        self.conn = sqlite3.connect(":memory:")
+        self.conn.row_factory = sqlite3.NamedRow
+
+    def tearDown(self):
+        self.conn.close()
+
+    def CheckInstance(self):
+        """row_factory creates NamedRow instances"""
+        row = self.conn.execute("SELECT 1, 2").fetchone()
+        self.assertIsInstance(row, sqlite3.NamedRow)
+        self.assertIn("NamedRow", row.__class__.__name__)
+        self.assertIn("NamedRow", repr(row))
+        self.assertIn(
+            "optimized row_factory for column name access", sqlite3.NamedRow.__doc__
+        )
+
+    def CheckByNumberIndex(self):
+        """Get values by numbered index [0]"""
+        row = self.conn.execute("SELECT 1, 2").fetchone()
+        # len(row)
+        self.assertEqual(2, len(row))
+
+        self.assertEqual(row[0], 1)
+        self.assertEqual(row[1], 2)
+        self.assertEqual(row[-1], 2)
+        self.assertEqual(row[-2], 1)
+
+        # Check past boundary
+        with self.assertRaises(IndexError):
+            row[len(row) + 1]
+        with self.assertRaises(IndexError):
+            row[-1 * (len(row) + 1)]
+
+    def CheckByStringIndex(self):
+        """Get values by string index ['field']"""
+        row = self.conn.execute('SELECT 1 AS a, 2 AS abcdefg, 4 AS "四"').fetchone()
+        self.assertEqual(1, row["a"])
+        self.assertEqual(2, row["abcdefg"])
+        self.assertEqual(4, row["四"])
+
+    def CheckByAttribute(self):
+        """Get values by attribute row.field"""
+        row = self.conn.execute('SELECT 1 AS a, 2 AS abcdefg, 4 AS "四"').fetchone()
+        self.assertEqual(1, row.a)
+        self.assertEqual(2, row.abcdefg)
+        self.assertEqual(4, row.四)
+
+    def CheckContainsField(self):
+        row = self.conn.execute("SELECT 1 AS a, 2 AS b").fetchone()
+        self.assertIn("a", row)
+        self.assertNotIn("A", row)
+
+    def CheckSlice(self):
+        """Does NamedRow slice like a list."""
+        row = self.conn.execute("SELECT 1, 2, 3, 4").fetchone()
+        self.assertEqual(row[0:0], ())
+        self.assertEqual(row[0:1], (1,))
+        self.assertEqual(row[1:3], (2, 3))
+        self.assertEqual(row[3:1], ())
+        # Explicit bounds are optional.
+        self.assertEqual(row[1:], (2, 3, 4))
+        self.assertEqual(row[:3], (1, 2, 3))
+        # Slices can use negative indices.
+        self.assertEqual(row[-2:-1], (3,))
+        self.assertEqual(row[-2:], (3, 4))
+        # Slicing supports steps.
+        self.assertEqual(row[0:4:2], (1, 3))
+        self.assertEqual(row[3:0:-2], (4, 2))
+
+    def CheckHashComparison(self):
+        row1 = self.conn.execute("SELECT 1 AS a, 2 AS b").fetchone()
+        row2 = self.conn.execute("SELECT 1 AS a, 2 AS b").fetchone()
+        row3 = self.conn.execute("SELECT 1 AS a, 3 AS b").fetchone()
+
+        self.assertEqual(row1, row1)
+        self.assertEqual(row1, row2)
+        self.assertTrue(row2 != row3)
+
+        self.assertFalse(row1 != row1)
+        self.assertFalse(row1 != row2)
+        self.assertFalse(row2 == row3)
+
+        self.assertEqual(row1, row2)
+        self.assertEqual(hash(row1), hash(row2))
+        self.assertNotEqual(row1, row3)
+        self.assertNotEqual(hash(row1), hash(row3))
+
+        with self.assertRaises(TypeError):
+            row1 > row2
+        with self.assertRaises(TypeError):
+            row1 < row2
+        with self.assertRaises(TypeError):
+            row1 >= row2
+        with self.assertRaises(TypeError):
+            row1 <= row1
+
+    def CheckFakeCursorClass(self):
+        # Issue #24257: Incorrect use of PyObject_IsInstance() caused
+        # segmentation fault.
+        # Issue #27861: Also applies for cursor factory.
+        class FakeCursor(str):
+            __class__ = sqlite3.Cursor
+
+        self.assertRaises(TypeError, self.conn.cursor, FakeCursor)
+        self.assertRaises(TypeError, sqlite3.NamedRow, FakeCursor(), ())
+
+    def CheckIterable(self):
+        """Is NamedRow iterable."""
+        row = self.conn.execute("SELECT 1 AS a, 2 AS abcdefg").fetchone()
+        self.assertEqual(2, len(row))
+        for col in row:
+            # Its is a key/value pair
+            self.assertEqual(2, len(col))
+
+    def CheckIterablePairs(self):
+        expected = (("a", 1), ("b", 2))
+        row = self.conn.execute("SELECT 1 AS a, 2 AS b").fetchone()
+        self.assertSequenceEqual(expected, [x for x in row])
+        self.assertSequenceEqual(expected, tuple(row))
+        self.assertSequenceEqual(expected, list(row))
+        self.assertSequenceEqual(dict(expected), dict(row))
+
+    def CheckExpectedExceptions(self):
+        row = self.conn.execute("SELECT 1 AS a").fetchone()
+        with self.assertRaises(IndexError):
+            row["no_such_field"]
+        with self.assertRaises(AttributeError):
+            row.no_such_field
+        with self.assertRaises(TypeError):
+            row[0] = 100
+        with self.assertRaises(TypeError):
+            row["a"] = 100
+        with self.assertRaises(TypeError) as err:
+            row.a = 100
+        self.assertIn("does not support item assignment", str(err.exception))
+
+        with self.assertRaises(TypeError):
+            setattr(row, "a", 100)
+
+
+def suite():
+    return unittest.makeSuite(NamedRowFactoryTests, "Check")
+
+
+def test():
+    runner = unittest.TextTestRunner()
+    runner.run(suite())
+
+
+if __name__ == "__main__":
+    unittest.main()

Lib/test/test_sqlite.py

@@ -1,26 +1,46 @@
 import test.support
 from test.support import import_helper
 
-# Skip test if _sqlite3 module not installed
+# Skip test if _sqlite3 module not ins
 import_helper.import_module('_sqlite3')
 
 import unittest
 import sqlite3
-from sqlite3.test import (dbapi, types, userfunctions,
-                                factory, transactions, hooks, regression,
-                                dump, backup)
+from sqlite3.test import (
+    dbapi,
+    types,
+    userfunctions,
+    factory,
+    transactions,
+    hooks,
+    regression,
+    dump,
+    backup,
+    test_namedrow,
+)
+
 
 def load_tests(*args):
     if test.support.verbose:
-        print("test_sqlite: testing with version",
-              "{!r}, sqlite_version {!r}".format(sqlite3.version,
-                                                 sqlite3.sqlite_version))
-    return unittest.TestSuite([dbapi.suite(), types.suite(),
-                               userfunctions.suite(),
-                               factory.suite(), transactions.suite(),
-                               hooks.suite(), regression.suite(),
-                               dump.suite(),
-                               backup.suite()])
+        print(
+            "test_sqlite: testing with version",
+            "{!r}, sqlite_version {!r}".format(sqlite3.version, sqlite3.sqlite_version),
+        )
+    return unittest.TestSuite(
+        [
+            dbapi.suite(),
+            types.suite(),
+            userfunctions.suite(),
+            factory.suite(),
+            transactions.suite(),
+            hooks.suite(),
+            regression.suite(),
+            dump.suite(),
+            backup.suite(),
+            test_namedrow.suite(),
+        ]
+    )
+
 
 if __name__ == "__main__":
     unittest.main()

Misc/ACKS

@@ -792,6 +792,7 @@ Manuel Jacob
 David Jacobs
 Kevin Jacobs
 Kjetil Jacobsen
+Clinton James
 Shantanu Jain
 Bertrand Janin
 Geert Jansen

Misc/NEWS.d/next/Library/2019-12-31-14-45-27.bpo-39170.bh3u0Q.rst

@@ -0,0 +1 @@
++ Add sqlite3 row_factory for attribute access with the same memory requirements and speed of sqlite3.Row