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

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

.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