Ability to truely use pyformat style

README.md

@@ -47,3 +47,108 @@ async with async_connection.cursor() as cursor:
     rows = await cursor.fetchmany(size=5)
     rows = await cursor.fetchall()
 ```
+
+## Query parameters
+
+### Standard mode (`pyformat=True`)
+
+Pass `pyformat=True` to `connect()` to enable familiar Python DB-API
+parameter syntax such as `%(name)s` and `%s`. This mode is opt-in: the
+default connection mode (`pyformat=False`) uses YDB-style `$name`
+placeholders instead, so `%(name)s` and `%s` do not work unless
+`pyformat=True` is set explicitly. The driver will convert placeholders
+and infer YDB types from Python values automatically.
+
+**Named parameters** — `%(name)s` with a `dict`:
+
+```python
+connection = ydb_dbapi.connect(
+    host="localhost", port="2136", database="/local",
+    pyformat=True,
+)
+
+with connection.cursor() as cursor:
+    cursor.execute(
+        "SELECT * FROM users WHERE id = %(id)s AND active = %(active)s",
+        {"id": 42, "active": True},
+    )
+```
+
+**Positional parameters** — `%s` with a `list` or `tuple`:
+
+```python
+with connection.cursor() as cursor:
+    cursor.execute(
+        "INSERT INTO users (id, name, score) VALUES (%s, %s, %s)",
+        [1, "Alice", 9.8],
+    )
+```
+
+Use `%%` to insert a literal `%` character in the query.
+
+The driver validates `pyformat` placeholders before executing the query:
+- do not mix named `%(name)s` and positional `%s` placeholders in one query;
+- named placeholders require a `dict` with bare keys like `{"id": 1}`;
+- positional placeholders require a `list` or `tuple`;
+- missing or extra parameters raise `ProgrammingError`;
+- keys starting with `$` are not allowed in `pyformat=True` mode.
+
+**Automatic type mapping:**
+
+| Python type        | YDB type    |
+|--------------------|-------------|
+| `bool`             | `Bool`      |
+| `int`              | `Int64`     |
+| `float`            | `Double`    |
+| `str`              | `Utf8`      |
+| `bytes`            | `String`    |
+| `datetime.datetime`| `Timestamp` |
+| `datetime.date`    | `Date`      |
+| `datetime.timedelta`| `Interval` |
+| `decimal.Decimal`  | `Decimal(22, 9)` |
+| `None`             | `NULL` (passed as-is) |
+
+**Explicit types with `ydb.TypedValue`:**
+
+When automatic inference is not suitable (e.g. you need `Int32` instead of
+`Int64`, or `Json`), wrap the value in `ydb.TypedValue` — it will be passed
+through unchanged:
+
+```python
+import ydb
+
+with connection.cursor() as cursor:
+    cursor.execute(
+        "INSERT INTO events (id, payload) VALUES (%(id)s, %(payload)s)",
+        {
+            "id": ydb.TypedValue(99, ydb.PrimitiveType.Int32),
+            "payload": ydb.TypedValue('{"key": "value"}', ydb.PrimitiveType.Json),
+        },
+    )
+```
+
+If the driver cannot infer a YDB type for a Python value, it raises
+`TypeError`. Use `ydb.TypedValue` for such values or when you need an explicit
+YDB type.
+
+### Native YDB mode (default, deprecated)
+
+> **Deprecated.** Native YDB mode is the current default for backwards
+> compatibility, but it will be removed in a future release. Migrate to
+> `pyformat=True` at your earliest convenience.
+
+By default (`pyformat=False`) the driver passes the query and parameters
+directly to the YDB SDK without any transformation. Use `$name` placeholders
+in the query and supply a `dict` with `$`-prefixed keys:
+
+```python
+connection = ydb_dbapi.connect(
+    host="localhost", port="2136", database="/local",
+)
+
+with connection.cursor() as cursor:
+    cursor.execute(
+        "SELECT * FROM users WHERE id = $id",
+        {"$id": ydb.TypedValue(42, ydb.PrimitiveType.Int64)},
+    )
+```

tests/test_connections.py

@@ -131,6 +131,47 @@ def _test_cursor_raw_query(self, connection: dbapi.Connection) -> None:
 
         maybe_await(cur.close())
 
+    def _test_cursor_pyformat_query(
+        self,
+        connection: dbapi.Connection,
+    ) -> None:
+        cur = connection.cursor()
+
+        with suppress(dbapi.DatabaseError):
+            maybe_await(cur.execute_scheme("DROP TABLE test_pyformat"))
+
+        maybe_await(
+            cur.execute_scheme(
+                """
+                CREATE TABLE test_pyformat(
+                    id Int64 NOT NULL,
+                    text Utf8,
+                    PRIMARY KEY (id)
+                )
+                """
+            )
+        )
+
+        maybe_await(
+            cur.execute(
+                "UPSERT INTO test_pyformat(id, text) VALUES (%s, %s)",
+                [17, "seventeen"],
+            )
+        )
+        maybe_await(
+            cur.execute(
+                "SELECT text FROM test_pyformat WHERE id = %(id)s",
+                {"id": 17},
+            )
+        )
+
+        row = cur.fetchone()
+        assert row is not None
+        assert row[0] == "seventeen"
+
+        maybe_await(cur.execute_scheme("DROP TABLE test_pyformat"))
+        maybe_await(cur.close())
+
     def _test_errors(
         self,
         connection: dbapi.Connection,
@@ -431,6 +472,13 @@ def test_connection(self, connection: dbapi.Connection) -> None:
     def test_cursor_raw_query(self, connection: dbapi.Connection) -> None:
         self._test_cursor_raw_query(connection)
 
+    def test_cursor_pyformat_query(self, connection_kwargs: dict) -> None:
+        connection = dbapi.connect(**{**connection_kwargs, "pyformat": True})
+        try:
+            self._test_cursor_pyformat_query(connection)
+        finally:
+            connection.close()
+
     def test_errors(self, connection: dbapi.Connection) -> None:
         self._test_errors(connection)
 
@@ -542,6 +590,19 @@ async def test_cursor_raw_query(
     ) -> None:
         await greenlet_spawn(self._test_cursor_raw_query, connection)
 
+    @pytest.mark.asyncio
+    async def test_cursor_pyformat_query(
+        self,
+        connection_kwargs: dict,
+    ) -> None:
+        connection = await dbapi.async_connect(
+            **{**connection_kwargs, "pyformat": True}
+        )
+        try:
+            await greenlet_spawn(self._test_cursor_pyformat_query, connection)
+        finally:
+            await connection.close()
+
     @pytest.mark.asyncio
     async def test_errors(self, connection: dbapi.AsyncConnection) -> None:
         await greenlet_spawn(self._test_errors, connection)