mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
yym
eb587fce7b
Use Hive-compatible identifier handling and template defaults so generated SQL returns real column values, while also fixing table schema parsing and adding required Hive runtime dependencies. Made-with: Cursor
54 lines
2.2 KiB
Python
54 lines
2.2 KiB
Python
# Author: Junjun
|
|
# Date: 2025/7/16
|
|
|
|
from enum import Enum
|
|
from typing import List
|
|
|
|
from common.utils.utils import equals_ignore_case
|
|
|
|
|
|
class ConnectType(Enum):
|
|
sqlalchemy = ('sqlalchemy')
|
|
py_driver = ('py_driver')
|
|
|
|
def __init__(self, type_name):
|
|
self.type_name = type_name
|
|
|
|
|
|
class DB(Enum):
|
|
excel = ('excel', 'Excel/CSV', '"', '"', ConnectType.sqlalchemy, 'PostgreSQL', [])
|
|
redshift = ('redshift', 'AWS Redshift', '"', '"', ConnectType.py_driver, 'AWS_Redshift', [])
|
|
ck = ('ck', 'ClickHouse', '"', '"', ConnectType.sqlalchemy, 'ClickHouse', [])
|
|
dm = ('dm', '达梦', '"', '"', ConnectType.py_driver, 'DM', [])
|
|
doris = ('doris', 'Apache Doris', '`', '`', ConnectType.py_driver, 'Doris', [])
|
|
es = ('es', 'Elasticsearch', '"', '"', ConnectType.py_driver, 'Elasticsearch', [])
|
|
kingbase = ('kingbase', 'Kingbase', '"', '"', ConnectType.py_driver, 'Kingbase', [])
|
|
sqlServer = ('sqlServer', 'Microsoft SQL Server', '[', ']', ConnectType.sqlalchemy, 'Microsoft_SQL_Server', [])
|
|
mysql = ('mysql', 'MySQL', '`', '`', ConnectType.sqlalchemy, 'MySQL', ['local_infile'])
|
|
oracle = ('oracle', 'Oracle', '"', '"', ConnectType.sqlalchemy, 'Oracle', [])
|
|
pg = ('pg', 'PostgreSQL', '"', '"', ConnectType.sqlalchemy, 'PostgreSQL', [])
|
|
starrocks = ('starrocks', 'StarRocks', '`', '`', ConnectType.py_driver, 'StarRocks', [])
|
|
sqlite = ('sqlite', 'SQLite', '"', '"', ConnectType.sqlalchemy, 'SQLite', [])
|
|
hive = ('hive', 'Apache Hive', '`', '`', ConnectType.py_driver, 'Hive', [])
|
|
|
|
def __init__(self, type, db_name, prefix, suffix, connect_type: ConnectType, template_name: str,
|
|
illegalParams: List[str]):
|
|
self.type = type
|
|
self.db_name = db_name
|
|
self.prefix = prefix
|
|
self.suffix = suffix
|
|
self.connect_type = connect_type
|
|
self.template_name = template_name
|
|
self.illegalParams = illegalParams
|
|
|
|
@classmethod
|
|
def get_db(cls, type, default_if_none=False):
|
|
for db in cls:
|
|
""" if db.type == type: """
|
|
if equals_ignore_case(db.type, type):
|
|
return db
|
|
if default_if_none:
|
|
return DB.pg
|
|
else:
|
|
raise ValueError(f"Invalid db type: {type}")
|