mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
e61fffc2ab
* drop pytz dependency and only make backports.zoneinfo required prior to the introduction of zoneinfo in the standard library because pytz docs mention that it shouldnt be required anymore and it is always good to cut back on uneeded 3rd party dependencies * some consistency updates * addressing review comments * don't mutate callers column_tzs * minor tweaks --------- Co-authored-by: Joe Spadola <[email protected]>
93 lines
4.7 KiB
Python
93 lines
4.7 KiB
Python
from collections.abc import Callable
|
|
from datetime import date, datetime
|
|
|
|
from clickhouse_connect.driver import Client
|
|
from clickhouse_connect.driver.binding import DT64Param
|
|
|
|
|
|
def test_params(param_client: Client, call, table_context: Callable):
|
|
result = call(param_client.query, "SELECT name, database FROM system.tables WHERE database = {db:String}", parameters={"db": "system"})
|
|
assert result.first_item["database"] == "system"
|
|
if param_client.min_version("21"):
|
|
result = call(
|
|
param_client.query,
|
|
"SELECT name, {col:String} FROM system.tables WHERE table ILIKE {t:String}",
|
|
parameters={"t": "%rr%", "col": "database"},
|
|
)
|
|
assert "rr" in result.first_item["name"]
|
|
|
|
first_date = datetime.strptime("Jun 1 2005 1:33PM", "%b %d %Y %I:%M%p").replace(tzinfo=param_client.server_tz)
|
|
second_date = datetime.strptime("Dec 25 2022 5:00AM", "%b %d %Y %I:%M%p").replace(tzinfo=param_client.server_tz)
|
|
with table_context("test_bind_params", ["key UInt64", "dt DateTime", "value String", "t Tuple(String, String)"]):
|
|
call(
|
|
param_client.insert,
|
|
"test_bind_params",
|
|
[
|
|
[1, first_date, "v11", ("one", "two")],
|
|
[2, second_date, "v21", ("t1", "t2")],
|
|
[3, datetime.now(), "v31", ("str1", "str2")],
|
|
],
|
|
)
|
|
result = call(param_client.query, "SELECT * FROM test_bind_params WHERE dt = {dt:DateTime}", parameters={"dt": second_date})
|
|
assert result.first_item["key"] == 2
|
|
result = call(param_client.query, "SELECT * FROM test_bind_params WHERE dt = %(dt)s", parameters={"dt": first_date})
|
|
assert result.first_item["key"] == 1
|
|
result = call(
|
|
param_client.query, "SELECT * FROM test_bind_params WHERE value != %(v)s AND value like '%%1'", parameters={"v": "v11"}
|
|
)
|
|
assert result.row_count == 2
|
|
result = call(param_client.query, "SELECT * FROM test_bind_params WHERE value IN %(tp)s", parameters={"tp": ("v18", "v31")})
|
|
assert result.first_item["key"] == 3
|
|
|
|
result = call(
|
|
param_client.query, "SELECT number FROM numbers(10) WHERE {n:Nullable(String)} IS NULL", parameters={"n": None}
|
|
).result_rows
|
|
assert len(result) == 10
|
|
|
|
date_params = [date(2023, 6, 1), date(2023, 8, 5)]
|
|
result = call(param_client.query, "SELECT {l:Array(Date)}", parameters={"l": date_params}).first_row
|
|
assert date_params == result[0]
|
|
|
|
dt_params = [datetime(2023, 6, 1, 7, 40, 2), datetime(2023, 8, 17, 20, 0, 10)]
|
|
result = call(param_client.query, "SELECT {l:Array(DateTime)}", parameters={"l": dt_params}).first_row
|
|
assert dt_params == result[0]
|
|
|
|
num_array_params = [2.5, 5.3, 7.4]
|
|
result = call(param_client.query, "SELECT {l:Array(Float64)}", parameters={"l": num_array_params}).first_row
|
|
assert num_array_params == result[0]
|
|
result = call(param_client.query, "SELECT %(l)s", parameters={"l": num_array_params}).first_row
|
|
assert num_array_params == result[0]
|
|
|
|
tp_params = ("str1", "str2")
|
|
result = call(param_client.query, "SELECT %(tp)s", parameters={"tp": tp_params}).first_row
|
|
assert tp_params == result[0]
|
|
|
|
num_params = {"p_0": 2, "p_1": 100523.55}
|
|
result = call(
|
|
param_client.query, "SELECT count() FROM system.tables WHERE total_rows > %(p_0)d and total_rows < %(p_1)f", parameters=num_params
|
|
)
|
|
assert result.first_row[0] > 0
|
|
|
|
|
|
def test_datetime_64_params(param_client: Client, call):
|
|
dt_values = [datetime(2023, 6, 1, 7, 40, 2, 250306), datetime(2023, 8, 17, 20, 0, 10, 777722)]
|
|
dt_params = {f"d{ix}": DT64Param(v) for ix, v in enumerate(dt_values)}
|
|
result = call(param_client.query, "SELECT {d0:DateTime64(3)}, {d1:Datetime64(9)}", parameters=dt_params).first_row
|
|
assert result[0] == dt_values[0].replace(microsecond=250000)
|
|
assert result[1] == dt_values[1]
|
|
|
|
result = call(param_client.query, "SELECT {a1:Array(DateTime64(6))}", parameters={"a1": [dt_params["d0"], dt_params["d1"]]}).first_row
|
|
assert result[0] == dt_values
|
|
|
|
dt_params = {f"d{ix}_64": v for ix, v in enumerate(dt_values)}
|
|
result = call(param_client.query, "SELECT {d0:DateTime64(3)}, {d1:Datetime64(9)}", parameters=dt_params).first_row
|
|
assert result[0] == dt_values[0].replace(microsecond=250000)
|
|
assert result[1] == dt_values[1]
|
|
|
|
result = call(param_client.query, "SELECT {a1:Array(DateTime64(6))}", parameters={"a1_64": dt_values}).first_row
|
|
assert result[0] == dt_values
|
|
|
|
dt_params = [DT64Param(v) for v in dt_values]
|
|
result = call(param_client.query, "SELECT %s as string, toDateTime64(%s,6) as dateTime", parameters=dt_params).first_row
|
|
assert result == ("2023-06-01 07:40:02.250306", dt_values[1])
|