mirror of
https://github.com/hanzo-ds/connect.git
synced 2026-10-30 20:31:24 +00:00
fdc88c02d3
* Fix: Wrap SQL strings with text() for SQLAlchemy 2.x compatibility * add tests * tweaks * update changelog * formatting --------- Co-authored-by: Joe Spadola <[email protected]>
73 lines
3.0 KiB
Python
73 lines
3.0 KiB
Python
import sqlalchemy as db
|
|
from sqlalchemy import inspect, text
|
|
from sqlalchemy.engine import Engine
|
|
|
|
from clickhouse_connect import common
|
|
from clickhouse_connect.cc_sqlalchemy.datatypes.sqltypes import Point, SimpleAggregateFunction, UInt32
|
|
|
|
|
|
def test_basic_reflection(test_engine: Engine):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
with test_engine.begin() as conn:
|
|
metadata = db.MetaData(schema="system")
|
|
table = db.Table("tables", metadata, autoload_with=test_engine)
|
|
query = db.select(table.columns.create_table_query)
|
|
result = conn.execute(query)
|
|
rows = result.fetchmany(100)
|
|
assert rows
|
|
|
|
|
|
def test_full_table_reflection(test_engine: Engine, test_db: str):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
with test_engine.begin() as conn:
|
|
conn.execute(text(f"DROP TABLE IF EXISTS {test_db}.reflect_test"))
|
|
conn.execute(
|
|
text(
|
|
f"CREATE TABLE {test_db}.reflect_test (key UInt32, value FixedString(20),"
|
|
+ "agg SimpleAggregateFunction(anyLast, String))"
|
|
+ "ENGINE AggregatingMergeTree ORDER BY key"
|
|
)
|
|
)
|
|
metadata = db.MetaData(schema=test_db)
|
|
table = db.Table("reflect_test", metadata, autoload_with=test_engine)
|
|
assert table.columns.key.type.__class__ == UInt32
|
|
assert table.columns.agg.type.__class__ == SimpleAggregateFunction
|
|
assert "MergeTree" in table.engine.name
|
|
|
|
|
|
def test_types_reflection(test_engine: Engine, test_db: str):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
with test_engine.begin() as conn:
|
|
conn.execute(text(f"DROP TABLE IF EXISTS {test_db}.sqlalchemy_types_test"))
|
|
conn.execute(text(f"CREATE TABLE {test_db}.sqlalchemy_types_test (key UInt32, pt Point) ENGINE MergeTree ORDER BY key"))
|
|
metadata = db.MetaData(schema=test_db)
|
|
table = db.Table("sqlalchemy_types_test", metadata, autoload_with=test_engine)
|
|
assert table.columns.key.type.__class__ == UInt32
|
|
assert table.columns.pt.type.__class__ == Point
|
|
assert "MergeTree" in table.engine.name
|
|
|
|
|
|
def test_table_exists(test_engine: Engine):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
inspector = inspect(test_engine)
|
|
assert inspector.has_table(table_name="columns", schema="system")
|
|
assert not inspector.has_table(table_name="nope", schema="fake_db")
|
|
|
|
|
|
def test_get_schema_names(test_engine: Engine):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
inspector = inspect(test_engine)
|
|
schema_names = inspector.get_schema_names()
|
|
assert isinstance(schema_names, list)
|
|
assert "system" in schema_names
|
|
assert "fake_db" not in schema_names
|
|
|
|
|
|
def test_get_table_names(test_engine: Engine, test_db: str):
|
|
common.set_setting("invalid_setting_action", "drop")
|
|
inspector = inspect(test_engine)
|
|
system_tables = inspector.get_table_names(schema="system")
|
|
assert isinstance(system_tables, list)
|
|
assert "columns" in system_tables
|
|
assert "fake_table" not in system_tables
|