Own datetime conversions at the DBAPI layer

Moves the wire-layer datetime handling down into the driver, matching PEP
249 (dates/times return datetime objects) and the convention used by
psycopg/mysqlclient/asyncpg. The wire codec now hands us raw ISO8601
strings and UNIXTIME int64s; this layer turns them into datetime objects
on read and stringifies datetime/date bind params on write.

types.py adds private helpers:
- _iso8601_from_datetime: formats datetime/date as the space-separated
  ISO layout the Go and C clients use. Accepts both naive and aware;
  naive formats without offset, aware preserves the offset.
- _datetime_from_iso8601: parses the inverse. Naive string → naive
  datetime; aware string → aware. No silent UTC assumption.
- _datetime_from_unixtime: int64 epoch seconds → UTC-aware datetime.
- _convert_bind_param: wraps the outbound datetime/date → str step.

cursor.py and aio/cursor.py:
- Read path: switched from query_raw to query_raw_typed so we know the
  per-column wire type code. A per-column converter table maps ISO8601
  and UNIXTIME values through the helpers above. Other types pass
  through unchanged (the wire codec already produced the right Python
  primitive).
- Write path: _convert_bind_param runs over params before they hit the
  wire encoder, which now rejects non-primitive inputs.
- description[i][1] is now populated with the wire ValueType integer
  instead of None — it's the natural type_code for PEP 249.

tests/integration/test_datetime_conversion.py exercises the contracts
end-to-end against a live cluster: naive-stays-naive, aware round-trips
with its original offset (+05:30 verified), microseconds preserved,
NULL decodes to None, datetime/date bind params reach the server,
description carries the correct wire type codes, UNIXTIME-tagged columns
(INTEGER values in DATETIME-typed columns, per dqlite server's query.c)
decode to datetime, and the AsyncCursor path goes through the same
conversion layer.

Three mock-based unit tests updated to patch query_raw_typed instead of
query_raw; the new method returns a 3-tuple (names, types, rows) where
the old one returned 2.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: antoineleclair

src/dqlitedbapi/aio/cursor.py

@@ -3,6 +3,7 @@
 from collections.abc import Sequence
 from typing import TYPE_CHECKING, Any
 
+from dqlitedbapi.cursor import _convert_params, _convert_row
 from dqlitedbapi.exceptions import InterfaceError
 
 if TYPE_CHECKING:
@@ -33,7 +34,7 @@ class AsyncCursor:
 
     def __init__(self, connection: "AsyncConnection") -> None:
         self._connection = connection
-        self._description: list[tuple[str, None, None, None, None, None, None]] | None = None
+        self._description: list[tuple[str, int | None, None, None, None, None, None]] | None = None
         self._rowcount = -1
         self._arraysize = 1
         self._rows: list[tuple[Any, ...]] = []
@@ -44,7 +45,7 @@ def __init__(self, connection: "AsyncConnection") -> None:
     @property
     def description(
         self,
-    ) -> list[tuple[str, None, None, None, None, None, None]] | None:
+    ) -> list[tuple[str, int | None, None, None, None, None, None]] | None:
         """Column descriptions for the last query."""
         return self._description
 
@@ -76,29 +77,40 @@ async def execute(
     ) -> "AsyncCursor":
         """Execute a database operation (query or command).
 
-        Routes through DqliteConnection's public API (execute/query_raw)
+        Routes through DqliteConnection's public API (execute/query_raw_typed)
         which goes through _run_protocol(), providing the _in_use guard,
         connection invalidation on fatal errors, and leader-change detection.
         The _op_lock serializes operations on the same connection.
         """
         self._check_closed()
 
         conn = await self._connection._ensure_connection()
-        params = list(parameters) if parameters is not None else None
+        params = _convert_params(parameters)
 
         # Determine if this is a query that returns rows.
         # Note: WITH ... INSERT/UPDATE/DELETE (without RETURNING) will be
-        # misrouted to query_raw. This is a known limitation of the heuristic.
+        # misrouted to query_raw_typed. This is a known limitation of the heuristic.
         normalized = _strip_leading_comments(operation).upper()
         is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 
         async with self._connection._op_lock:
             if is_query:
-                columns, rows = await conn.query_raw(operation, params)
-                self._description = [(name, None, None, None, None, None, None) for name in columns]
-                self._rows = [tuple(row) for row in rows]
+                columns, column_types, rows = await conn.query_raw_typed(operation, params)
+                self._description = [
+                    (
+                        name,
+                        column_types[i] if i < len(column_types) else None,
+                        None,
+                        None,
+                        None,
+                        None,
+                        None,
+                    )
+                    for i, name in enumerate(columns)
+                ]
+                self._rows = [_convert_row(row, column_types) for row in rows]
                 self._row_index = 0
                 self._rowcount = len(rows)
             else:

src/dqlitedbapi/cursor.py

@@ -1,14 +1,47 @@
 """PEP 249 Cursor implementation for dqlite."""
 
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 from typing import TYPE_CHECKING, Any
 
+from dqlitewire.constants import ValueType
+
 from dqlitedbapi.exceptions import InterfaceError
+from dqlitedbapi.types import (
+    _convert_bind_param,
+    _datetime_from_iso8601,
+    _datetime_from_unixtime,
+)
 
 if TYPE_CHECKING:
     from dqlitedbapi.connection import Connection
 
 
+# Per-wire-type result converters. NULL/empty values pass through as None;
+# unrecognized types pass through unchanged (the wire codec already produced
+# an appropriate Python primitive).
+_RESULT_CONVERTERS: dict[int, Callable[[Any], Any]] = {
+    int(ValueType.ISO8601): lambda v: _datetime_from_iso8601(v) if isinstance(v, str) else v,
+    int(ValueType.UNIXTIME): lambda v: _datetime_from_unixtime(v) if isinstance(v, int) else v,
+}
+
+
+def _convert_row(row: Sequence[Any], column_types: Sequence[int]) -> tuple[Any, ...]:
+    """Apply result-side converters to a row based on its column wire types."""
+    result = list(row)
+    for i, tcode in enumerate(column_types):
+        converter = _RESULT_CONVERTERS.get(tcode)
+        if converter is not None and result[i] is not None:
+            result[i] = converter(result[i])
+    return tuple(result)
+
+
+def _convert_params(params: Sequence[Any] | None) -> list[Any] | None:
+    """Convert driver-level bind parameters (e.g. datetime) to wire primitives."""
+    if params is None:
+        return None
+    return [_convert_bind_param(p) for p in params]
+
+
 def _strip_leading_comments(sql: str) -> str:
     """Strip leading SQL comments (-- and /* */) and whitespace."""
     s = sql.strip()
@@ -33,7 +66,7 @@ class Cursor:
 
     def __init__(self, connection: "Connection") -> None:
         self._connection = connection
-        self._description: list[tuple[str, None, None, None, None, None, None]] | None = None
+        self._description: list[tuple[str, int | None, None, None, None, None, None]] | None = None
         self._rowcount = -1
         self._arraysize = 1
         self._rows: list[tuple[Any, ...]] = []
@@ -44,13 +77,15 @@ def __init__(self, connection: "Connection") -> None:
     @property
     def description(
         self,
-    ) -> list[tuple[str, None, None, None, None, None, None]] | None:
+    ) -> list[tuple[str, int | None, None, None, None, None, None]] | None:
         """Column descriptions for the last query.
 
         Returns a list of 7-tuples:
         (name, type_code, display_size, internal_size, precision, scale, null_ok)
 
-        Only name is populated; others are None for compatibility.
+        ``type_code`` is the wire-level ``ValueType`` integer from the first
+        result frame (e.g. 10 for ISO8601, 9 for UNIXTIME). The other fields
+        are None — dqlite doesn't expose them.
         """
         return self._description
 
@@ -91,25 +126,36 @@ def execute(self, operation: str, parameters: Sequence[Any] | None = None) -> "C
     async def _execute_async(self, operation: str, parameters: Sequence[Any] | None = None) -> None:
         """Async implementation of execute.
 
-        Routes through DqliteConnection's public API (execute/query_raw)
+        Routes through DqliteConnection's public API (execute/query_raw_typed)
         which goes through _run_protocol(), providing the _in_use guard,
         connection invalidation on fatal errors, and leader-change detection.
         """
         conn = await self._connection._get_async_connection()
-        params = list(parameters) if parameters is not None else None
+        params = _convert_params(parameters)
 
         # Determine if this is a query that returns rows.
         # Note: WITH ... INSERT/UPDATE/DELETE (without RETURNING) will be
-        # misrouted to query_raw. This is a known limitation of the heuristic.
+        # misrouted to query_raw_typed. This is a known limitation of the heuristic.
         normalized = _strip_leading_comments(operation).upper()
         is_query = normalized.startswith(("SELECT", "PRAGMA", "EXPLAIN", "WITH")) or (
             " RETURNING " in normalized or normalized.endswith(" RETURNING")
         )
 
         if is_query:
-            columns, rows = await conn.query_raw(operation, params)
-            self._description = [(name, None, None, None, None, None, None) for name in columns]
-            self._rows = [tuple(row) for row in rows]
+            columns, column_types, rows = await conn.query_raw_typed(operation, params)
+            self._description = [
+                (
+                    name,
+                    column_types[i] if i < len(column_types) else None,
+                    None,
+                    None,
+                    None,
+                    None,
+                    None,
+                )
+                for i, name in enumerate(columns)
+            ]
+            self._rows = [_convert_row(row, column_types) for row in rows]
             self._row_index = 0
             self._rowcount = len(rows)
         else:

src/dqlitedbapi/types.py

@@ -1,6 +1,7 @@
 """PEP 249 type objects and constructors for dqlite."""
 
 import datetime
+from typing import Any
 
 
 # Type constructors
@@ -62,3 +63,82 @@ def __hash__(self) -> int:
 NUMBER = _DBAPIType("INTEGER", "INT", "SMALLINT", "BIGINT", "REAL", "FLOAT", "DOUBLE", "NUMERIC")
 DATETIME = _DBAPIType("DATE", "TIME", "TIMESTAMP", "DATETIME")
 ROWID = _DBAPIType("ROWID", "INTEGER PRIMARY KEY")
+
+
+# Internal conversion helpers.
+#
+# The wire codec deals only in primitives (ISO8601 → str, UNIXTIME → int64).
+# PEP 249 specifies that drivers SHOULD return datetime objects for date/time
+# columns — and every major Python driver (psycopg, mysqlclient, asyncpg, ...)
+# does. These helpers implement that conversion at the driver (DBAPI) layer,
+# matching Go's database/sql driver split.
+
+
+def _iso8601_from_datetime(value: datetime.datetime | datetime.date) -> str:
+    """Format a datetime/date as an ISO 8601 string for wire transmission.
+
+    Uses the space-separated layout so values are byte-for-byte comparable
+    with what Go and the C client produce. Accepts both naive and
+    timezone-aware datetimes — naive values round-trip as naive (matching
+    pysqlite semantics), aware values preserve the offset.
+    """
+    if isinstance(value, datetime.datetime):
+        base = f"{value.year:04d}" + value.strftime("-%m-%d %H:%M:%S")
+        if value.microsecond:
+            base += f".{value.microsecond:06d}"
+        if value.tzinfo is None:
+            return base
+        offset = value.utcoffset()
+        assert offset is not None
+        total_seconds = int(offset.total_seconds())
+        sign = "+" if total_seconds >= 0 else "-"
+        hours, remainder = divmod(abs(total_seconds), 3600)
+        minutes = remainder // 60
+        return base + f"{sign}{hours:02d}:{minutes:02d}"
+    # datetime.date (must come after datetime check — datetime is a subclass).
+    return value.isoformat()
+
+
+def _datetime_from_iso8601(text: str) -> datetime.datetime | None:
+    """Parse an ISO 8601 string into ``datetime.datetime``.
+
+    Returns ``None`` for the empty string — pre-null-patch dqlite servers
+    sometimes emit empty text for NULL datetime cells, and the modern
+    server still tolerates empty ISO8601 values. Returning None matches
+    PEP 249 NULL semantics.
+
+    Naive input round-trips as naive; aware input preserves the offset.
+    """
+    if not text:
+        return None
+    s = text[:-1] + "+00:00" if text.endswith("Z") else text
+    try:
+        return datetime.datetime.fromisoformat(s)
+    except ValueError:
+        pass
+    try:
+        d = datetime.date.fromisoformat(s)
+    except ValueError as exc:
+        raise ValueError(f"Cannot parse ISO 8601 datetime: {text!r}") from exc
+    return datetime.datetime(d.year, d.month, d.day)
+
+
+def _datetime_from_unixtime(value: int) -> datetime.datetime:
+    """Decode a UNIXTIME int64 into a UTC-aware ``datetime.datetime``.
+
+    UNIXTIME is unambiguously seconds-since-epoch in UTC, so returning a
+    UTC-aware value is faithful. Callers that want local time can convert.
+    """
+    return datetime.datetime.fromtimestamp(value, tz=datetime.UTC)
+
+
+def _convert_bind_param(value: Any) -> Any:
+    """Map driver-level Python types to wire primitives.
+
+    The wire codec accepts only bool/int/float/str/bytes/None; datetime and
+    date are driver-level conveniences that we stringify to ISO 8601 before
+    handing off. Everything else passes through unchanged.
+    """
+    if isinstance(value, datetime.datetime | datetime.date):
+        return _iso8601_from_datetime(value)
+    return value