fix(parse-colnames): parse colnames from [decltype] in py>=3.7

Respect detect_type to PARSE_COLNAMES before parsing [decltype] from column aliases

Author: danielebriggi

.devcontainer/py3.7/devcontainer.json

@@ -0,0 +1,38 @@
+// For format details, see https://aka.ms/devcontainer.json. For config options, see the
+// README at: https://github.com/devcontainers/templates/tree/main/src/python
+{
+	"name": "Python 3.7",
+	// Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
+	"image": "mcr.microsoft.com/devcontainers/python:3.7",
+	"features": {
+		"ghcr.io/warrenbuckley/codespace-features/sqlite:1": {}
+	},
+
+	// Features to add to the dev container. More info: https://containers.dev/features.
+	// "features": {},
+
+	// Use 'forwardPorts' to make a list of ports inside the container available locally.
+	// "forwardPorts": [],
+
+	// Use 'postCreateCommand' to run commands after the container is created.
+	// "postCreateCommand": "pip3 install --user -r requirements.txt",
+
+	// Configure tool-specific properties.
+	"customizations": {
+		"vscode": {
+			"extensions": [
+				"littlefoxteam.vscode-python-test-adapter",
+				"jkillian.custom-local-formatters",
+				"ms-python.vscode-pylance",
+				"ms-python.python",
+				"ms-python.debugpy",
+				"ms-python.black-formatter",
+				"ms-python.isort",
+				"ms-toolsai.jupyter"
+			]
+		}
+	}
+
+	// Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
+	// "remoteUser": "root"
+}

.gitignore

@@ -2,7 +2,7 @@
 build/
 experiments/
 sdk/
-venv/
+venv*/
 main.dSYM/
 .env
 *.pyc

src/sqlitecloud/dbapi2.py

@@ -5,6 +5,7 @@
 #
 import logging
 import re
+import sys
 from datetime import date, datetime
 from typing import (
     Any,
@@ -370,11 +371,23 @@ def description(
         if not self._is_result_rowset():
             return None
 
+        # Since py3.7:
+        # bpo-39652: The column name found in sqlite3.Cursor.description is
+        # now truncated on the first ‘[’ only if the PARSE_COLNAMES option is set.
+        # https://github.com/python/cpython/issues/83833
+        parse_colname = (
+            self.connection.detect_types & PARSE_COLNAMES
+        ) == PARSE_COLNAMES
+        if sys.version_info < (3, 7):
+            parse_colname = True
+
         description = ()
         for i in range(self._resultset.ncols):
+            colname = self._resultset.colname[i]
+
             description += (
                 (
-                    self._parse_colname(self._resultset.colname[i])[0],
+                    self._parse_colname(colname)[0] if parse_colname else colname,
                     None,
                     None,
                     None,
@@ -550,28 +563,6 @@ def setinputsizes(self, sizes) -> None:
     def setoutputsize(self, size, column=None) -> None:
         pass
 
-    def _parse_colname(self, colname: str) -> Tuple[str, str]:
-        """
-        Parse the column name to extract the column name and the
-        declared type if present when it follows the syntax `colname [decltype]`.
-
-        Args:
-            colname (str): The column name with optional declared type.
-                Eg: "mycol [mytype]"
-
-        Returns:
-            Tuple[str, str]: The column name and the declared type.
-                Eg: ("mycol", "mytype")
-        """
-        # search for `[mytype]` in `mycol [mytype]`
-        pattern = r"\[(.*?)\]"
-
-        matches = re.findall(pattern, colname)
-        if not matches or len(matches) == 0:
-            return colname, None
-
-        return colname.replace(f"[{matches[0]}]", "").strip(), matches[0]
-
     def _call_row_factory(self, row: Tuple) -> object:
         if self.row_factory is None:
             return row
@@ -634,6 +625,28 @@ def _convert_value(
 
         return value
 
+    def _parse_colname(self, colname: str) -> Tuple[str, str]:
+        """
+        Parse the column name to extract the column name and the
+        declared type if present when it follows the syntax `colname [decltype]`.
+
+        Args:
+            colname (str): The column name with optional declared type.
+                Eg: "mycol [mytype]"
+
+        Returns:
+            Tuple[str, str]: The column name and the declared type.
+                Eg: ("mycol", "mytype")
+        """
+        # search for `[mytype]` in `mycol [mytype]`
+        pattern = r"\[(.*?)\]"
+
+        matches = re.findall(pattern, colname)
+        if not matches or len(matches) == 0:
+            return colname, None
+
+        return colname.replace(f"[{matches[0]}]", "").strip(), matches[0]
+
     def _parse_colnames(self, value: Any, colname: str) -> Optional[Any]:
         """Convert the value using the explicit type in the column name."""
         _, decltype = self._parse_colname(colname)

src/tests/integration/test_sqlite3_parity.py

@@ -1,5 +1,6 @@
 import random
 import sqlite3
+import sys
 import time
 from datetime import date, datetime
 
@@ -226,17 +227,85 @@ def test_description(self, connection, request):
             "sqlite3_connection",
         ],
     )
-    def test_cursor_description_with_explicit_decltype(self, connection, request):
+    @pytest.mark.parametrize(
+        "value",
+        [
+            ("'hello world'", "'hello world'"),
+            ('"hello" "world"', "world"),
+            ('"hello" "my world"', "my world"),
+        ],
+    )
+    def test_cursor_description_with_column_alias(self, connection, value, request):
         connection = request.getfixturevalue(connection)
 
-        cursor = connection.execute(
-            'SELECT "hello world", "hello" as "my world [sphere]", "hello" "world", "hello" "my world"'
-        )
+        cursor = connection.execute(f"SELECT {value[0]}")
+
+        assert cursor.description[0][0] == value[1]
+
+    # Only for py3.6
+    @pytest.mark.skipif(
+        sys.version_info >= (3, 7), reason="Different behavior in py>=3.7"
+    )
+    @pytest.mark.parametrize(
+        "connection",
+        [
+            "sqlitecloud_dbapi2_connection",
+            "sqlite3_connection",
+        ],
+    )
+    @pytest.mark.parametrize(
+        "value",
+        [
+            ('"hello" as "world [sphere]"', "world"),
+            ('"hello" as "my world [sphere]"', "my world"),
+            ('"hello" "world [sphere]"', "world"),
+        ],
+    )
+    def test_cursor_description_with_explicit_decltype_regardless_detect_type(
+        self, connection, value, request
+    ):
+        """In py3.6 the `[decltype]` in column name is always parsed regardless the PARSE_COLNAMES.
+        See bpo-39652 https://github.com/python/cpython/issues/83833"""
+
+        connection = request.getfixturevalue(connection)
+
+        cursor = connection.execute(f"SELECT {value[0]}")
+
+        assert cursor.description[0][0] == value[1]
+
+    # Only for py>=3.7
+    @pytest.mark.skipif(sys.version_info < (3, 7), reason="Different behavior in py3.6")
+    @pytest.mark.parametrize(
+        "connection, module",
+        [
+            (get_sqlitecloud_dbapi2_connection, sqlitecloud),
+            (get_sqlite3_connection, sqlite3),
+        ],
+    )
+    @pytest.mark.parametrize(
+        "value, expected, parse_colnames",
+        [
+            ('"hello" as "world [sphere]"', "world", True),
+            ('"hello" as "my world [sphere]"', "my world", True),
+            ('"hello" "world [sphere]"', "world", True),
+            ('"hello" as "world [sphere]"', "world [sphere]", False),
+            ('"hello" as "my world [sphere]"', "my world [sphere]", False),
+            ('"hello" "world [sphere]"', "world [sphere]", False),
+        ],
+    )
+    def test_cursor_description_with_explicit_decltype(
+        self, connection, module, value, expected, parse_colnames
+    ):
+        """Since py3.7 the parsed of `[decltype]` disabled when PARSE_COLNAMES.
+        See bpo-39652 https://github.com/python/cpython/issues/83833"""
+        if parse_colnames:
+            connection = next(connection(module.PARSE_COLNAMES))
+        else:
+            connection = next(connection())
+
+        cursor = connection.execute(f"SELECT {value}")
 
-        assert cursor.description[0][0] == '"hello world"'
-        assert cursor.description[1][0] == "my world"
-        assert cursor.description[2][0] == "world"
-        assert cursor.description[3][0] == "my world"
+        assert cursor.description[0][0] == expected
 
     def test_fetch_one(self, sqlitecloud_dbapi2_connection, sqlite3_connection):
         sqlitecloud_connection = sqlitecloud_dbapi2_connection