mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
Joe SpadolaandGitHub
fa4d66308c
* comprehensive join support * fix for wrong values table function construction * fix final and sample override bug * support join using, and array join with multiple cols * adjustments for sqa1.4 compat * consolidate readme * address PR comments
64 lines
2.0 KiB
Python
64 lines
2.0 KiB
Python
import pytest
|
|
import sqlalchemy as db
|
|
from sqlalchemy.engine import Engine
|
|
|
|
from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import DateTime
|
|
|
|
SA_2 = db.__version__ >= "2"
|
|
|
|
|
|
def test_values_round_trip_multi_column(test_engine: Engine):
|
|
with test_engine.begin() as conn:
|
|
values_clause = db.values(
|
|
db.column("id", db.Integer),
|
|
db.column("name", db.String),
|
|
name="v",
|
|
).data([(17, "user_1"), (29, "user_2")])
|
|
|
|
rows = conn.execute(
|
|
db.select(values_clause.c.id, values_clause.c.name).select_from(values_clause).order_by(values_clause.c.id)
|
|
).fetchall()
|
|
|
|
assert [(row.id, row.name) for row in rows] == [(17, "user_1"), (29, "user_2")]
|
|
|
|
|
|
def test_values_round_trip_single_column(test_engine: Engine):
|
|
with test_engine.begin() as conn:
|
|
values_clause = db.values(
|
|
db.column("score", db.Integer),
|
|
name="v",
|
|
).data([(17,), (29,)])
|
|
|
|
total = conn.execute(db.select(db.func.sum(values_clause.c.score)).select_from(values_clause)).scalar()
|
|
|
|
assert total == 46
|
|
|
|
|
|
def test_values_round_trip_type_name_with_quotes(test_engine: Engine):
|
|
with test_engine.begin() as conn:
|
|
values_clause = db.values(
|
|
db.column("event_ts", DateTime("UTC")),
|
|
name="v",
|
|
).data([("2024-01-02 03:04:05",)])
|
|
|
|
value = conn.execute(db.select(values_clause.c.event_ts).select_from(values_clause)).scalar()
|
|
|
|
assert str(value).startswith("2024-01-02 03:04:05")
|
|
|
|
|
|
@pytest.mark.skipif(not SA_2, reason="Values.cte() was added in SA 2.x")
|
|
def test_values_cte_round_trip(test_engine: Engine):
|
|
with test_engine.begin() as conn:
|
|
values_clause = (
|
|
db.values(
|
|
db.column("id", db.Integer),
|
|
name="v",
|
|
)
|
|
.data([(17,), (29,)])
|
|
.cte("input_rows")
|
|
)
|
|
|
|
value = conn.execute(db.select(db.func.max(values_clause.c.id)).select_from(values_clause)).scalar()
|
|
|
|
assert value == 29
|