feat(parse-columns): support to PARSE_COLUMNS

Column names are parsed to extract the explicit decltype.

Author: danielebriggi

src/sqlitecloud/__init__.py

@@ -7,7 +7,9 @@
     PARSE_DECLTYPES,
     Connection,
     Cursor,
+    adapters,
     connect,
+    converters,
     register_adapter,
     register_converter,
 )
@@ -21,6 +23,8 @@
     "register_converter",
     "PARSE_DECLTYPES",
     "PARSE_COLNAMES",
+    "adapters",
+    "converters",
 ]
 
 VERSION = "0.0.79"

src/sqlitecloud/dbapi2.py

@@ -4,6 +4,7 @@
 # https://peps.python.org/pep-0249/
 #
 import logging
+import re
 from datetime import date, datetime
 from typing import (
     Any,
@@ -24,6 +25,7 @@
     SQLiteCloudConfig,
     SQLiteCloudConnect,
     SQLiteCloudException,
+    SQLiteDataTypes,
 )
 from sqlitecloud.driver import Driver
 from sqlitecloud.resultset import (
@@ -51,9 +53,9 @@
 PARSE_COLNAMES = 2
 
 # Adapters registry to convert Python types to SQLite types
-_adapters = {}
+adapters: Dict[Type[Any], Callable[[Any], SQLiteDataTypes]] = {}
 # Converters registry to convert SQLite types to Python types
-_converters = {}
+converters: Dict[str, Callable[[bytes], Any]] = {}
 
 
 @overload
@@ -148,8 +150,8 @@ def register_adapter(
         callable (Callable): The callable that converts the type into a supported
             SQLite supported type.
     """
-    global _adapters
-    _adapters[pytype] = adapter_callable
+    registry = _get_adapters_registry()
+    registry[pytype] = adapter_callable
 
 
 def register_converter(type_name: str, converter: Callable[[bytes], Any]) -> None:
@@ -161,8 +163,16 @@ def register_converter(type_name: str, converter: Callable[[bytes], Any]) -> Non
             The match with the name of the type in the query is case-insensitive.
         converter (Callable): The callable that converts the bytestring into the custom Python type.
     """
-    global _converters
-    _converters[type_name.lower()] = converter
+    registry = _get_converters_registry()
+    registry[type_name.lower()] = converter
+
+
+def _get_adapters_registry() -> dict:
+    return adapters
+
+
+def _get_converters_registry() -> dict:
+    return converters
 
 
 class Connection:
@@ -307,8 +317,8 @@ def _apply_adapter(self, value: Any) -> SQLiteTypes:
         Returns:
             SQLiteTypes: The SQLite supported type or the given value when no adapter is found.
         """
-        if type(value) in _adapters:
-            return _adapters[type(value)](value)
+        if type(value) in adapters:
+            return adapters[type(value)](value)
 
         if hasattr(value, "__conform__"):
             # we don't support sqlite3.PrepareProtocol
@@ -365,7 +375,7 @@ def description(
         for i in range(self._resultset.ncols):
             description += (
                 (
-                    self._resultset.colname[i],
+                    self._parse_colname(self._resultset.colname[i])[0],
                     None,
                     None,
                     None,
@@ -541,6 +551,28 @@ 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
@@ -572,11 +604,26 @@ def _adapt_parameters(self, parameters: Union[Dict, Tuple]) -> Union[Dict, Tuple
 
         return tuple(self._connection._apply_adapter(p) for p in parameters)
 
-    def _convert_value(self, value: Any, decltype: Optional[str]) -> Any:
-        # todo: parse columns first
+    def _convert_value(
+        self, value: Any, colname: Optional[str], decltype: Optional[str]
+    ) -> Any:
+        if (
+            colname
+            and (self.connection.detect_types & PARSE_COLNAMES) == PARSE_COLNAMES
+        ):
+            try:
+                return self._parse_colnames(value, colname)
+            except MissingDecltypeException:
+                pass
 
-        if (self.connection.detect_types & PARSE_DECLTYPES) == PARSE_DECLTYPES:
-            return self._parse_decltypes(value, decltype)
+        if (
+            decltype
+            and (self.connection.detect_types & PARSE_DECLTYPES) == PARSE_DECLTYPES
+        ):
+            try:
+                return self._parse_decltypes(value, decltype)
+            except MissingDecltypeException:
+                pass
 
         if decltype == SQLITECLOUD_VALUE_TYPE.TEXT.value or (
             decltype is None and isinstance(value, str)
@@ -585,16 +632,27 @@ def _convert_value(self, value: Any, decltype: Optional[str]) -> Any:
 
         return value
 
-    def _parse_decltypes(self, value: Any, decltype: str) -> Any:
+    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)
+
+        if decltype:
+            return self._parse_decltypes(value, decltype)
+
+        raise MissingDecltypeException(f"No decltype declared for: {decltype}")
+
+    def _parse_decltypes(self, value: Any, decltype: str) -> Optional[Any]:
+        """Convert the value by calling the registered converter for the given decltype."""
         decltype = decltype.lower()
-        if decltype in _converters:
+        registry = _get_converters_registry()
+        if decltype in registry:
             # sqlite3 always passes value as bytes
             value = (
                 str(value).encode("utf-8") if not isinstance(value, bytes) else value
             )
-            return _converters[decltype](value)
+            return registry[decltype](value)
 
-        return value
+        raise MissingDecltypeException(f"No decltype registered for: {decltype}")
 
     def _apply_text_factory(self, value: Any) -> Any:
         """Use Connection.text_factory to convert value with TEXT column or
@@ -614,9 +672,10 @@ def _get_value(self, row: int, col: int) -> Optional[Any]:
             return None
 
         value = self._resultset.get_value(row, col)
+        colname = self._resultset.get_name(col)
         decltype = self._resultset.get_decltype(col)
 
-        return self._convert_value(value, decltype)
+        return self._convert_value(value, colname, decltype)
 
     def __iter__(self) -> "Cursor":
         return self
@@ -638,6 +697,12 @@ def __next__(self) -> Optional[Tuple[Any]]:
         raise StopIteration
 
 
+class MissingDecltypeException(Exception):
+    def __init__(self, message: str) -> None:
+        super().__init__(message)
+        self.message = message
+
+
 def register_adapters_and_converters():
     """
     sqlite3 default adapters and converters.

src/tests/conftest.py

@@ -1,5 +1,6 @@
 import os
 import sqlite3
+from copy import deepcopy
 
 import pytest
 from dotenv import load_dotenv
@@ -14,6 +15,29 @@ def load_env_vars():
     load_dotenv(".env")
 
 
+@pytest.fixture(autouse=True)
+def reset_module_state():
+    original_sqlc_adapters = deepcopy(sqlitecloud.adapters)
+    original_sqlc_converters = deepcopy(sqlitecloud.converters)
+
+    original_sql_adapters = deepcopy(sqlite3.adapters)
+    original_sql_converters = deepcopy(sqlite3.converters)
+
+    yield
+
+    sqlitecloud.adapters.clear()
+    sqlitecloud.converters.clear()
+
+    sqlite3.adapters.clear()
+    sqlite3.converters.clear()
+
+    sqlitecloud.adapters.update(original_sqlc_adapters)
+    sqlitecloud.converters.update(original_sqlc_converters)
+
+    sqlite3.adapters.update(original_sql_adapters)
+    sqlite3.converters.update(original_sql_converters)
+
+
 @pytest.fixture()
 def sqlitecloud_connection():
     account = SQLiteCloudAccount()
@@ -42,7 +66,16 @@ def sqlitecloud_dbapi2_connection():
     # the test.
     # Fixtures are both cached and cannot be called
     # directly whithin the test.
-    yield next(get_sqlitecloud_dbapi2_connection())
+    connection_generator = get_sqlitecloud_dbapi2_connection()
+
+    connection = next(connection_generator)
+
+    yield connection
+
+    try:
+        next(connection_generator)
+    except StopIteration:
+        pass
 
 
 def get_sqlitecloud_dbapi2_connection(detect_types: int = 0):
@@ -62,6 +95,20 @@ def get_sqlitecloud_dbapi2_connection(detect_types: int = 0):
     connection.close()
 
 
+@pytest.fixture()
+def sqlite3_connection():
+    connection_generator = get_sqlite3_connection()
+
+    connection = next(connection_generator)
+
+    yield connection
+
+    try:
+        next(connection_generator)
+    except StopIteration:
+        pass
+
+
 def get_sqlite3_connection(detect_types: int = 0):
     # set isolation_level=None to enable autocommit
     # and to be aligned with the behavior of SQLite Cloud
@@ -71,4 +118,5 @@ def get_sqlite3_connection(detect_types: int = 0):
         detect_types=detect_types,
     )
     yield connection
+
     connection.close()