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]>
122 lines
3.8 KiB
Python
122 lines
3.8 KiB
Python
from sqlalchemy import text
|
|
from sqlalchemy.engine.default import DefaultDialect
|
|
|
|
from clickhouse_connect import dbapi
|
|
from clickhouse_connect.cc_sqlalchemy import dialect_name, ischema_names
|
|
from clickhouse_connect.cc_sqlalchemy.inspector import ChInspector
|
|
from clickhouse_connect.cc_sqlalchemy.sql import full_table
|
|
from clickhouse_connect.cc_sqlalchemy.sql.compiler import ChStatementCompiler
|
|
from clickhouse_connect.cc_sqlalchemy.sql.ddlcompiler import ChDDLCompiler
|
|
from clickhouse_connect.cc_sqlalchemy.sql.preparer import ChIdentifierPreparer
|
|
from clickhouse_connect.driver.binding import format_str, quote_identifier
|
|
|
|
|
|
class ClickHouseDialect(DefaultDialect):
|
|
"""
|
|
See :py:class:`sqlalchemy.engine.interfaces`
|
|
"""
|
|
|
|
name = dialect_name
|
|
driver = "connect"
|
|
|
|
default_schema_name = "default"
|
|
supports_native_decimal = True
|
|
supports_native_boolean = True
|
|
supports_statement_cache = False
|
|
returns_unicode_strings = True
|
|
postfetch_lastrowid = False
|
|
ddl_compiler = ChDDLCompiler
|
|
statement_compiler = ChStatementCompiler
|
|
preparer = ChIdentifierPreparer
|
|
description_encoding = None
|
|
max_identifier_length = 127
|
|
ischema_names = ischema_names
|
|
inspector = ChInspector
|
|
|
|
# SQA 1 compatibility
|
|
|
|
@classmethod
|
|
def dbapi(cls):
|
|
return dbapi
|
|
|
|
# SQA 2 compatibility
|
|
|
|
@classmethod
|
|
def import_dbapi(cls):
|
|
return dbapi
|
|
|
|
def initialize(self, connection):
|
|
pass
|
|
|
|
def get_schema_names(self, connection, **_):
|
|
return [row.name for row in connection.execute(text("SHOW DATABASES"))]
|
|
|
|
@staticmethod
|
|
def has_database(connection, db_name):
|
|
return (connection.execute(text(f"SELECT name FROM system.databases WHERE name = {format_str(db_name)}"))).rowcount > 0
|
|
|
|
def get_table_names(self, connection, schema=None, **kw):
|
|
cmd = "SHOW TABLES"
|
|
if schema:
|
|
cmd += " FROM " + quote_identifier(schema)
|
|
return [row.name for row in connection.execute(text(cmd))]
|
|
|
|
def get_primary_keys(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def get_pk_constraint(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def get_foreign_keys(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def get_temp_table_names(self, connection, schema=None, **kw):
|
|
return []
|
|
|
|
def get_view_names(self, connection, schema=None, **kw):
|
|
return []
|
|
|
|
def get_temp_view_names(self, connection, schema=None, **kw):
|
|
return []
|
|
|
|
def get_view_definition(self, connection, view_name, schema=None, **kw):
|
|
pass
|
|
|
|
def get_indexes(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def get_unique_constraints(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def get_check_constraints(self, connection, table_name, schema=None, **kw):
|
|
return []
|
|
|
|
def has_table(self, connection, table_name, schema=None, **_kw):
|
|
result = connection.execute(text(f"EXISTS TABLE {full_table(table_name, schema)}"))
|
|
row = result.fetchone()
|
|
return row[0] == 1
|
|
|
|
def has_sequence(self, connection, sequence_name, schema=None, **_kw):
|
|
return False
|
|
|
|
def do_begin_twophase(self, connection, xid):
|
|
raise NotImplementedError
|
|
|
|
def do_prepare_twophase(self, connection, xid):
|
|
raise NotImplementedError
|
|
|
|
def do_rollback_twophase(self, connection, xid, is_prepared=True, recover=False):
|
|
raise NotImplementedError
|
|
|
|
def do_commit_twophase(self, connection, xid, is_prepared=True, recover=False):
|
|
raise NotImplementedError
|
|
|
|
def do_recover_twophase(self, connection):
|
|
raise NotImplementedError
|
|
|
|
def set_isolation_level(self, dbapi_conn, level):
|
|
pass
|
|
|
|
def get_isolation_level(self, dbapi_conn):
|
|
return None
|