mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
14b4ca5773
* fix dbapi executemany crashing on sequence rows * align dbapi integration test fixture with cloud test settings --------- Co-authored-by: Joe S <[email protected]>
52 lines
2.0 KiB
Python
52 lines
2.0 KiB
Python
from collections.abc import Callable
|
|
|
|
from pytest import fixture
|
|
|
|
from clickhouse_connect import dbapi
|
|
from tests.integration_tests.conftest import TestConfig
|
|
|
|
|
|
@fixture(name="dbapi_connection")
|
|
def dbapi_connection_fixture(test_config: TestConfig, test_db: str):
|
|
settings = {}
|
|
if test_config.insert_quorum:
|
|
settings["insert_quorum"] = test_config.insert_quorum
|
|
elif test_config.cloud:
|
|
settings["select_sequential_consistency"] = 1
|
|
connection = dbapi.connect(
|
|
host=test_config.host,
|
|
port=test_config.port,
|
|
username=test_config.username,
|
|
password=test_config.password,
|
|
database=test_db,
|
|
compress=test_config.compress,
|
|
**settings,
|
|
)
|
|
yield connection
|
|
connection.close()
|
|
|
|
|
|
def test_executemany_with_tuple_rows(dbapi_connection, table_context: Callable):
|
|
"""Regression test: executemany with sequence rows (e.g. Airflow's
|
|
DbApiHook.insert_rows) used to crash with AttributeError in _try_bulk_insert.
|
|
"""
|
|
with table_context("dbapi_executemany_tuples", ["id UInt32", "name String"]):
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.executemany(
|
|
"INSERT INTO dbapi_executemany_tuples (id, name) VALUES (%s, %s)",
|
|
[(13, "user_1"), (79, "user_2")],
|
|
)
|
|
cursor.execute("SELECT id, name FROM dbapi_executemany_tuples ORDER BY id")
|
|
assert cursor.fetchall() == [(13, "user_1"), (79, "user_2")]
|
|
|
|
|
|
def test_executemany_with_dict_rows(dbapi_connection, table_context: Callable):
|
|
with table_context("dbapi_executemany_dicts", ["id UInt32", "name String"]):
|
|
cursor = dbapi_connection.cursor()
|
|
cursor.executemany(
|
|
"INSERT INTO dbapi_executemany_dicts (id, name) VALUES (%(id)s, %(name)s)",
|
|
[{"id": 13, "name": "user_1"}, {"id": 79, "name": "user_2"}],
|
|
)
|
|
cursor.execute("SELECT id, name FROM dbapi_executemany_dicts ORDER BY id")
|
|
assert cursor.fetchall() == [(13, "user_1"), (79, "user_2")]
|