feat(sqlalchemy): expose cycle-3 DoS knobs via URL + pin async dialect_description

Post-review follow-up. Adds two entries to the URL query allowlist:
- max_continuation_frames (int) — per-query frame cap (ISSUE-98)
- trust_server_heartbeat (bool) — opt-in server-heartbeat trust
  (ISSUE-101)

bool parsing uses a URL-friendly parser because bool("False") is
truthy. Accepts 1/true/yes/on as true; everything else as false.

The URL value converter now handles the str | tuple[str, ...] shape
that SQLAlchemy uses for multi-valued query params (takes the last
occurrence).

Also adds the async variant of the dialect_description pin test
(correctness review flagged the sync-only coverage).

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

Author: antoineleclair

src/sqlalchemydqlite/base.py

@@ -3,6 +3,7 @@
 import contextlib
 import datetime
 import warnings
+from collections.abc import Callable
 from typing import Any
 
 from sqlalchemy import types as sqltypes
@@ -111,7 +112,14 @@ def import_dbapi(cls) -> Any:
 
     # Whitelist of URL query parameters we forward to the DBAPI connect
     # call. Unknown keys raise ``ArgumentError`` so typos surface.
-    _URL_QUERY_ALLOWED: dict[str, type] = {"timeout": float, "max_total_rows": int}
+    # ``trust_server_heartbeat`` uses a URL-friendly bool parser because
+    # bool("False") evaluates truthy (non-empty string).
+    _URL_QUERY_ALLOWED: dict[str, Callable[[str], Any]] = {
+        "timeout": float,
+        "max_total_rows": int,
+        "max_continuation_frames": int,
+        "trust_server_heartbeat": lambda s: s.strip().lower() in ("1", "true", "yes", "on"),
+    }
 
     def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
         """Create connection arguments from URL.
@@ -137,11 +145,15 @@ def create_connect_args(self, url: URL) -> tuple[list[Any], dict[str, Any]]:
                     f"Allowed: {sorted(self._URL_QUERY_ALLOWED)}"
                 )
             converter = self._URL_QUERY_ALLOWED[key]
+            # URL query values can be str or tuple[str, ...] (when a key
+            # appears multiple times). Take the last occurrence.
+            raw_str = raw[-1] if isinstance(raw, tuple) else raw
             try:
-                kwargs[key] = converter(raw)
+                kwargs[key] = converter(raw_str)
             except (TypeError, ValueError) as e:
                 raise ArgumentError(
-                    f"Cannot convert URL query {key}={raw!r} to {converter.__name__}: {e}"
+                    f"Cannot convert URL query {key}={raw!r} to "
+                    f"{getattr(converter, '__name__', 'expected type')}: {e}"
                 ) from e
 
         return [], kwargs

tests/test_dialect.py

@@ -59,6 +59,11 @@ def test_dialect_description(self) -> None:
         # (ISSUE-89).
         assert DqliteDialect().dialect_description == "dqlite+dqlitedbapi"
 
+    def test_async_dialect_description(self) -> None:
+        # Mirror ISSUE-89 pin for the async dialect; review agent flagged
+        # that the sync test alone could mask an async-side drift.
+        assert DqliteDialect_aio().dialect_description == "dqlite+dqlitedbapi_aio"
+
 
 class TestDqliteDialectAio:
     def test_dialect_name(self) -> None:

tests/test_dialect_dialect_config.py

@@ -88,6 +88,44 @@ def test_timeout_and_max_total_rows_together(self) -> None:
         assert kwargs["timeout"] == 3.5
         assert kwargs["max_total_rows"] == 250
 
+    def test_max_continuation_frames_forwarded(self) -> None:
+        """ISSUE-98 URL plumbing — post-review follow-up."""
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db?max_continuation_frames=500")
+        _, kwargs = dialect.create_connect_args(url)
+        assert kwargs["max_continuation_frames"] == 500
+
+    def test_max_continuation_frames_rejects_non_int(self) -> None:
+        dialect = DqliteDialect()
+        url = make_url("dqlite://host:19001/db?max_continuation_frames=nope")
+        with pytest.raises(ArgumentError, match="int"):
+            dialect.create_connect_args(url)
+
+    @pytest.mark.parametrize(
+        "raw,expected",
+        [
+            ("1", True),
+            ("true", True),
+            ("True", True),
+            ("YES", True),
+            ("on", True),
+            ("0", False),
+            ("false", False),
+            ("no", False),
+            ("off", False),
+        ],
+    )
+    def test_trust_server_heartbeat_parses_boolean(self, raw: str, expected: bool) -> None:
+        """ISSUE-101 URL plumbing — post-review follow-up.
+
+        URL values arrive as strings; bool("False") would evaluate
+        truthy if used directly, so we use a dedicated parser.
+        """
+        dialect = DqliteDialect()
+        url = make_url(f"dqlite://host:19001/db?trust_server_heartbeat={raw}")
+        _, kwargs = dialect.create_connect_args(url)
+        assert kwargs["trust_server_heartbeat"] is expected
+
 
 class TestDoPingNarrowExceptions:
     def test_returns_true_on_success(self) -> None: