use bound params and format_str in DDL/inspector

CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Bug Fixes
 - Async client: retry once when a pooled keep-alive connection is closed by the server and aiohttp raises `ServerDisconnectedError` with the default `"Server disconnected"` message. The existing retry path covered `"Connection reset"` and `"Remote end closed"`, but not the bare `ServerDisconnectedError()` produced by recent aiohttp versions, which surfaced as an `OperationalError("Network Error: Server disconnected")` on the first request after an idle period.
 - SQLAlchemy `Bool` type now accepts and forwards `**kwargs` to the underlying `SqlaBoolean` constructor. SQLAlchemy's `SchemaType` machinery passes internal kwargs (e.g., `_create_events`) when copying or adapting the type during ORM model use or `Table.to_metadata()`, which previously raised a `TypeError`. Fixes [#705](https://github.com/ClickHouse/clickhouse-connect/issues/705)
+- SQLAlchemy: `CreateDatabase` with `engine="Replicated"` now emits a closing `)` after the `(zoo_path, shard, replica)` arguments, fixing previously invalid DDL on this path. The same arguments and the `system.tables` lookup in `get_engine` now go through bound parameters and the existing `format_str` helper instead of raw f-string interpolation.
 
 ## 1.0.0rc1, 2026-04-22
 

clickhouse_connect/cc_sqlalchemy/ddl/custom.py

@@ -1,7 +1,7 @@
 from sqlalchemy.exc import ArgumentError
 from sqlalchemy.sql.ddl import DDL
 
-from clickhouse_connect.driver.binding import quote_identifier
+from clickhouse_connect.driver.binding import format_str, quote_identifier
 
 
 class CreateDatabase(DDL):
@@ -33,7 +33,7 @@ def __init__(
             if engine == "Replicated":
                 if not zoo_path:
                     raise ArgumentError("zoo_path is required for Replicated Database Engine")
-                stmt += f" ('{zoo_path}', '{shard_name}', '{replica_name}'"
+                stmt += f" ({format_str(zoo_path)}, {format_str(shard_name)}, {format_str(replica_name)})"
         super().__init__(stmt)
 
 

clickhouse_connect/cc_sqlalchemy/inspector.py

@@ -12,7 +12,10 @@
 
 
 def get_engine(connection, table_name, schema=None):
-    result_set = connection.execute(text(f"SELECT engine_full FROM system.tables WHERE database = '{schema}' and name = '{table_name}'"))
+    result_set = connection.execute(
+        text("SELECT engine_full FROM system.tables WHERE database = :schema AND name = :table_name"),
+        {"schema": schema, "table_name": table_name},
+    )
     row = next(result_set, None)
     if not row:
         raise NoResultFound(f"Table {schema}.{table_name} does not exist")

tests/unit_tests/test_sqlalchemy/test_create_database.py

@@ -0,0 +1,69 @@
+import pytest
+from sqlalchemy.exc import ArgumentError
+
+from clickhouse_connect.cc_sqlalchemy.ddl.custom import CreateDatabase, DropDatabase
+
+
+def _ddl(stmt):
+    return stmt.statement if isinstance(stmt.statement, str) else str(stmt.statement)
+
+
+def test_create_database_plain():
+    assert _ddl(CreateDatabase("mydb")) == "CREATE DATABASE `mydb`"
+
+
+def test_create_database_if_not_exists():
+    assert _ddl(CreateDatabase("mydb", exists_ok=True)) == "CREATE DATABASE IF NOT EXISTS `mydb`"
+
+
+def test_create_database_atomic_engine():
+    assert _ddl(CreateDatabase("mydb", engine="Atomic")) == "CREATE DATABASE `mydb` Engine Atomic"
+
+
+def test_create_database_unknown_engine():
+    with pytest.raises(ArgumentError):
+        CreateDatabase("mydb", engine="Bogus")
+
+
+def test_create_database_replicated_requires_zoo_path():
+    with pytest.raises(ArgumentError):
+        CreateDatabase("mydb", engine="Replicated")
+
+
+def test_create_database_replicated_default_macros():
+    ddl = _ddl(CreateDatabase("mydb", engine="Replicated", zoo_path="/clickhouse/databases/mydb"))
+    assert ddl == ("CREATE DATABASE `mydb` Engine Replicated ('/clickhouse/databases/mydb', '{shard}', '{replica}')")
+
+
+def test_create_database_replicated_explicit_args():
+    ddl = _ddl(
+        CreateDatabase(
+            "mydb",
+            engine="Replicated",
+            zoo_path="/clickhouse/databases/mydb",
+            shard_name="shard_1",
+            replica_name="replica_a",
+        )
+    )
+    assert ddl == ("CREATE DATABASE `mydb` Engine Replicated ('/clickhouse/databases/mydb', 'shard_1', 'replica_a')")
+
+
+def test_create_database_replicated_escapes_quote():
+    ddl = _ddl(
+        CreateDatabase(
+            "mydb",
+            engine="Replicated",
+            zoo_path="/clickhouse/'evil",
+            shard_name="shard'1",
+            replica_name="replica\\a",
+        )
+    )
+    assert ddl == ("CREATE DATABASE `mydb` Engine Replicated ('/clickhouse/\\'evil', 'shard\\'1', 'replica\\\\a')")
+
+
+def test_drop_database_plain():
+    assert _ddl(DropDatabase("mydb")) == "DROP DATABASE `mydb`"
+
+
+def test_drop_database_if_exists():
+    assert _ddl(DropDatabase("mydb", missing_ok=True)) == "DROP DATABASE IF EXISTS `mydb`"