mirror of
https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git
synced 2026-11-08 01:06:59 +08:00
feat: support DM datasource
This commit is contained in:
@@ -2,12 +2,14 @@ import datetime
|
||||
import json
|
||||
from typing import List, Optional
|
||||
|
||||
import dmPython
|
||||
from fastapi import HTTPException
|
||||
from sqlalchemy import and_, text
|
||||
from sqlmodel import select
|
||||
|
||||
from apps.datasource.crud.permission import get_column_permission_fields, get_row_permission_filters, is_normal_user
|
||||
from apps.datasource.utils.utils import aes_decrypt
|
||||
from apps.db.constant import ConnectType
|
||||
from apps.db.constant import DB
|
||||
from apps.db.db import get_engine, get_tables, get_fields, exec_sql
|
||||
from apps.db.engine import get_engine_config, get_engine_conn
|
||||
@@ -45,16 +47,32 @@ def check_status_by_id(session: SessionDep, trans: Trans, ds_id: int, is_raise:
|
||||
|
||||
|
||||
def check_status(session: SessionDep, trans: Trans, ds: CoreDatasource, is_raise: bool = False):
|
||||
conn = get_engine(ds, 10)
|
||||
try:
|
||||
with conn.connect() as connection:
|
||||
SQLBotLogUtil.info("success")
|
||||
return True
|
||||
except Exception as e:
|
||||
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
|
||||
if is_raise:
|
||||
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
|
||||
return False
|
||||
db = DB.get_db(ds.type)
|
||||
if db.connect_type == ConnectType.sqlalchemy:
|
||||
conn = get_engine(ds, 10)
|
||||
try:
|
||||
with conn.connect() as connection:
|
||||
SQLBotLogUtil.info("success")
|
||||
return True
|
||||
except Exception as e:
|
||||
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
|
||||
if is_raise:
|
||||
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
|
||||
return False
|
||||
else:
|
||||
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration)))
|
||||
if ds.type == 'dm':
|
||||
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
|
||||
port=conf.port) as conn, conn.cursor() as cursor:
|
||||
try:
|
||||
cursor.execute('select 1').fetchall()
|
||||
SQLBotLogUtil.info("success")
|
||||
return True
|
||||
except Exception as e:
|
||||
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
|
||||
if is_raise:
|
||||
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
|
||||
return False
|
||||
|
||||
|
||||
def check_name(session: SessionDep, trans: Trans, user: CurrentUser, ds: CoreDatasource):
|
||||
@@ -159,7 +177,7 @@ def getFieldsByDs(session: SessionDep, ds: CoreDatasource, table_name: str):
|
||||
|
||||
def execSql(session: SessionDep, id: int, sql: str):
|
||||
ds = session.exec(select(CoreDatasource).where(CoreDatasource.id == id)).first()
|
||||
return exec_sql(ds, sql)
|
||||
return exec_sql(ds, sql, True)
|
||||
|
||||
|
||||
def sync_table(session: SessionDep, ds: CoreDatasource, tables: List[CoreTable]):
|
||||
@@ -297,7 +315,11 @@ def preview(session: SessionDep, current_user: CurrentUser, id: int, data: Table
|
||||
sql = f"""SELECT "{'", "'.join(fields)}" FROM "{data.table.table_name}"
|
||||
{where}
|
||||
LIMIT 100"""
|
||||
return exec_sql(ds, sql)
|
||||
elif ds.type == "dm":
|
||||
sql = f"""SELECT "{'", "'.join(fields)}" FROM "{conf.dbSchema}"."{data.table.table_name}"
|
||||
{where}
|
||||
LIMIT 100"""
|
||||
return exec_sql(ds, sql, True)
|
||||
|
||||
|
||||
def fieldEnum(session: SessionDep, id: int):
|
||||
@@ -313,7 +335,7 @@ def fieldEnum(session: SessionDep, id: int):
|
||||
|
||||
db = DB.get_db(ds.type)
|
||||
sql = f"""SELECT DISTINCT {db.prefix}{field.field_name}{db.suffix} FROM {db.prefix}{table.table_name}{db.suffix}"""
|
||||
res = exec_sql(ds, sql)
|
||||
res = exec_sql(ds, sql, True)
|
||||
return [item.get(res.get('fields')[0]) for item in res.get('data')]
|
||||
|
||||
|
||||
|
||||
@@ -4,18 +4,28 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class DB(Enum):
|
||||
mysql = ('mysql', '`', '`')
|
||||
sqlServer = ('sqlServer', '[', ']')
|
||||
pg = ('pg', '"', '"')
|
||||
excel = ('excel', '"', '"')
|
||||
oracle = ('oracle', '"', '"')
|
||||
ck = ('ClickHouse', '"', '"')
|
||||
class ConnectType(Enum):
|
||||
sqlalchemy = ('sqlalchemy')
|
||||
py_driver = ('py_driver')
|
||||
|
||||
def __init__(self, type, prefix, suffix):
|
||||
def __init__(self, type_name):
|
||||
self.type_name = type_name
|
||||
|
||||
|
||||
class DB(Enum):
|
||||
mysql = ('mysql', '`', '`', ConnectType.sqlalchemy)
|
||||
sqlServer = ('sqlServer', '[', ']', ConnectType.sqlalchemy)
|
||||
pg = ('pg', '"', '"', ConnectType.sqlalchemy)
|
||||
excel = ('excel', '"', '"', ConnectType.sqlalchemy)
|
||||
oracle = ('oracle', '"', '"', ConnectType.sqlalchemy)
|
||||
ck = ('ck', '"', '"', ConnectType.sqlalchemy)
|
||||
dm = ('dm', '"', '"', ConnectType.py_driver)
|
||||
|
||||
def __init__(self, type, prefix, suffix, connect_type: ConnectType):
|
||||
self.type = type
|
||||
self.prefix = prefix
|
||||
self.suffix = suffix
|
||||
self.connect_type = connect_type
|
||||
|
||||
@classmethod
|
||||
def get_db(cls, type):
|
||||
|
||||
+244
-180
@@ -3,11 +3,13 @@ import json
|
||||
import urllib.parse
|
||||
from decimal import Decimal
|
||||
|
||||
import dmPython
|
||||
from sqlalchemy import create_engine, text, Engine
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from apps.datasource.models.datasource import DatasourceConf, CoreDatasource, TableSchema, ColumnSchema
|
||||
from apps.datasource.utils.utils import aes_decrypt
|
||||
from apps.db.constant import DB, ConnectType
|
||||
from apps.db.engine import get_engine_config
|
||||
from apps.system.crud.assistant import get_ds_engine
|
||||
from apps.system.schemas.system_schema import AssistantOutDsSchema
|
||||
@@ -86,189 +88,251 @@ def get_session(ds: CoreDatasource | AssistantOutDsSchema):
|
||||
|
||||
def get_tables(ds: CoreDatasource):
|
||||
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config()
|
||||
with get_session(ds) as session:
|
||||
sql: str = ''
|
||||
if ds.type == "mysql":
|
||||
sql = f"""
|
||||
SELECT
|
||||
TABLE_NAME,
|
||||
TABLE_COMMENT
|
||||
FROM
|
||||
information_schema.TABLES
|
||||
WHERE
|
||||
TABLE_SCHEMA = '{conf.database}'
|
||||
"""
|
||||
elif ds.type == "sqlServer":
|
||||
sql = f"""
|
||||
SELECT
|
||||
TABLE_NAME AS [TABLE_NAME],
|
||||
ISNULL(ep.value, '') AS [TABLE_COMMENT]
|
||||
FROM
|
||||
INFORMATION_SCHEMA.TABLES t
|
||||
LEFT JOIN
|
||||
sys.extended_properties ep
|
||||
ON ep.major_id = OBJECT_ID(t.TABLE_SCHEMA + '.' + t.TABLE_NAME)
|
||||
AND ep.minor_id = 0
|
||||
AND ep.name = 'MS_Description'
|
||||
WHERE
|
||||
t.TABLE_TYPE IN ('BASE TABLE', 'VIEW')
|
||||
AND t.TABLE_SCHEMA = '{conf.dbSchema}'
|
||||
"""
|
||||
elif ds.type == "pg" or ds.type == "excel":
|
||||
sql = """
|
||||
SELECT c.relname AS TABLE_NAME,
|
||||
COALESCE(d.description, obj_description(c.oid)) AS TABLE_COMMENT
|
||||
FROM pg_class c
|
||||
LEFT JOIN
|
||||
pg_namespace n ON n.oid = c.relnamespace
|
||||
LEFT JOIN
|
||||
pg_description d ON d.objoid = c.oid AND d.objsubid = 0
|
||||
WHERE n.nspname = current_schema()
|
||||
AND c.relkind IN ('r', 'v', 'p', 'm')
|
||||
AND c.relname NOT LIKE 'pg_%'
|
||||
AND c.relname NOT LIKE 'sql_%'
|
||||
ORDER BY c.relname \
|
||||
"""
|
||||
elif ds.type == "oracle":
|
||||
sql = f"""
|
||||
SELECT
|
||||
t.TABLE_NAME AS "TABLE_NAME",
|
||||
NVL(c.COMMENTS, '') AS "TABLE_COMMENT"
|
||||
FROM (
|
||||
SELECT TABLE_NAME, 'TABLE' AS OBJECT_TYPE
|
||||
FROM DBA_TABLES
|
||||
WHERE OWNER = '{conf.dbSchema}'
|
||||
UNION ALL
|
||||
SELECT VIEW_NAME AS TABLE_NAME, 'VIEW' AS OBJECT_TYPE
|
||||
FROM DBA_VIEWS
|
||||
WHERE OWNER = '{conf.dbSchema}'
|
||||
) t
|
||||
LEFT JOIN DBA_TAB_COMMENTS c
|
||||
ON t.TABLE_NAME = c.TABLE_NAME
|
||||
AND c.TABLE_TYPE = t.OBJECT_TYPE
|
||||
AND c.OWNER = '{conf.dbSchema}'
|
||||
ORDER BY t.TABLE_NAME
|
||||
"""
|
||||
elif ds.type == "ck":
|
||||
sql = f"""
|
||||
SELECT name, engine
|
||||
FROM system.tables
|
||||
WHERE database = '{conf.database}'
|
||||
AND engine NOT IN ('Dictionary')
|
||||
ORDER BY name
|
||||
"""
|
||||
with session.execute(text(sql)) as result:
|
||||
res = result.fetchall()
|
||||
res_list = [TableSchema(*item) for item in res]
|
||||
return res_list
|
||||
db = DB.get_db(ds.type)
|
||||
if db.connect_type == ConnectType.sqlalchemy:
|
||||
with get_session(ds) as session:
|
||||
sql: str = ''
|
||||
if ds.type == "mysql":
|
||||
sql = f"""
|
||||
SELECT
|
||||
TABLE_NAME,
|
||||
TABLE_COMMENT
|
||||
FROM
|
||||
information_schema.TABLES
|
||||
WHERE
|
||||
TABLE_SCHEMA = '{conf.database}'
|
||||
"""
|
||||
elif ds.type == "sqlServer":
|
||||
sql = f"""
|
||||
SELECT
|
||||
TABLE_NAME AS [TABLE_NAME],
|
||||
ISNULL(ep.value, '') AS [TABLE_COMMENT]
|
||||
FROM
|
||||
INFORMATION_SCHEMA.TABLES t
|
||||
LEFT JOIN
|
||||
sys.extended_properties ep
|
||||
ON ep.major_id = OBJECT_ID(t.TABLE_SCHEMA + '.' + t.TABLE_NAME)
|
||||
AND ep.minor_id = 0
|
||||
AND ep.name = 'MS_Description'
|
||||
WHERE
|
||||
t.TABLE_TYPE IN ('BASE TABLE', 'VIEW')
|
||||
AND t.TABLE_SCHEMA = '{conf.dbSchema}'
|
||||
"""
|
||||
elif ds.type == "pg" or ds.type == "excel":
|
||||
sql = """
|
||||
SELECT c.relname AS TABLE_NAME,
|
||||
COALESCE(d.description, obj_description(c.oid)) AS TABLE_COMMENT
|
||||
FROM pg_class c
|
||||
LEFT JOIN
|
||||
pg_namespace n ON n.oid = c.relnamespace
|
||||
LEFT JOIN
|
||||
pg_description d ON d.objoid = c.oid AND d.objsubid = 0
|
||||
WHERE n.nspname = current_schema()
|
||||
AND c.relkind IN ('r', 'v', 'p', 'm')
|
||||
AND c.relname NOT LIKE 'pg_%'
|
||||
AND c.relname NOT LIKE 'sql_%'
|
||||
ORDER BY c.relname \
|
||||
"""
|
||||
elif ds.type == "oracle":
|
||||
sql = f"""
|
||||
SELECT
|
||||
t.TABLE_NAME AS "TABLE_NAME",
|
||||
NVL(c.COMMENTS, '') AS "TABLE_COMMENT"
|
||||
FROM (
|
||||
SELECT TABLE_NAME, 'TABLE' AS OBJECT_TYPE
|
||||
FROM DBA_TABLES
|
||||
WHERE OWNER = '{conf.dbSchema}'
|
||||
UNION ALL
|
||||
SELECT VIEW_NAME AS TABLE_NAME, 'VIEW' AS OBJECT_TYPE
|
||||
FROM DBA_VIEWS
|
||||
WHERE OWNER = '{conf.dbSchema}'
|
||||
) t
|
||||
LEFT JOIN DBA_TAB_COMMENTS c
|
||||
ON t.TABLE_NAME = c.TABLE_NAME
|
||||
AND c.TABLE_TYPE = t.OBJECT_TYPE
|
||||
AND c.OWNER = '{conf.dbSchema}'
|
||||
ORDER BY t.TABLE_NAME
|
||||
"""
|
||||
elif ds.type == "ck":
|
||||
sql = f"""
|
||||
SELECT name, comment
|
||||
FROM system.tables
|
||||
WHERE database = '{conf.database}'
|
||||
AND engine NOT IN ('Dictionary')
|
||||
ORDER BY name
|
||||
"""
|
||||
with session.execute(text(sql)) as result:
|
||||
res = result.fetchall()
|
||||
res_list = [TableSchema(*item) for item in res]
|
||||
return res_list
|
||||
else:
|
||||
if ds.type == 'dm':
|
||||
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
|
||||
port=conf.port) as conn, conn.cursor() as cursor:
|
||||
cursor.execute(f"""select table_name, comments
|
||||
from all_tab_comments
|
||||
where owner='{conf.dbSchema}'
|
||||
AND (table_type = 'TABLE' or table_type = 'VIEW')
|
||||
""")
|
||||
res = cursor.fetchall()
|
||||
res_list = [TableSchema(*item) for item in res]
|
||||
return res_list
|
||||
|
||||
|
||||
def get_fields(ds: CoreDatasource, table_name: str = None):
|
||||
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config()
|
||||
with get_session(ds) as session:
|
||||
sql: str = ''
|
||||
if ds.type == "mysql":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
COLUMN_NAME,
|
||||
DATA_TYPE,
|
||||
COLUMN_COMMENT
|
||||
FROM
|
||||
INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE
|
||||
TABLE_SCHEMA = '{conf.database}'
|
||||
"""
|
||||
sql2 = f" AND TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "sqlServer":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
COLUMN_NAME AS [COLUMN_NAME],
|
||||
DATA_TYPE AS [DATA_TYPE],
|
||||
ISNULL(EP.value, '') AS [COLUMN_COMMENT]
|
||||
FROM
|
||||
INFORMATION_SCHEMA.COLUMNS C
|
||||
LEFT JOIN
|
||||
sys.extended_properties EP
|
||||
ON EP.major_id = OBJECT_ID(C.TABLE_SCHEMA + '.' + C.TABLE_NAME)
|
||||
AND EP.minor_id = C.ORDINAL_POSITION
|
||||
AND EP.name = 'MS_Description'
|
||||
WHERE
|
||||
C.TABLE_SCHEMA = '{conf.dbSchema}'
|
||||
"""
|
||||
sql2 = f" AND C.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "pg" or ds.type == "excel":
|
||||
sql1 = """
|
||||
SELECT a.attname AS COLUMN_NAME,
|
||||
pg_catalog.format_type(a.atttypid, a.atttypmod) AS DATA_TYPE,
|
||||
col_description(c.oid, a.attnum) AS COLUMN_COMMENT
|
||||
FROM pg_catalog.pg_attribute a
|
||||
JOIN
|
||||
pg_catalog.pg_class c ON a.attrelid = c.oid
|
||||
JOIN
|
||||
pg_catalog.pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE n.nspname = current_schema()
|
||||
AND a.attnum > 0
|
||||
AND NOT a.attisdropped \
|
||||
"""
|
||||
sql2 = f" AND c.relname = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "oracle":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
col.COLUMN_NAME AS "COLUMN_NAME",
|
||||
(CASE
|
||||
WHEN col.DATA_TYPE IN ('VARCHAR2', 'CHAR', 'NVARCHAR2', 'NCHAR')
|
||||
THEN col.DATA_TYPE || '(' || col.DATA_LENGTH || ')'
|
||||
WHEN col.DATA_TYPE = 'NUMBER' AND col.DATA_PRECISION IS NOT NULL
|
||||
THEN col.DATA_TYPE || '(' || col.DATA_PRECISION ||
|
||||
CASE WHEN col.DATA_SCALE > 0 THEN ',' || col.DATA_SCALE END || ')'
|
||||
ELSE col.DATA_TYPE
|
||||
END) AS "DATA_TYPE",
|
||||
NVL(com.COMMENTS, '') AS "COLUMN_COMMENT"
|
||||
FROM
|
||||
DBA_TAB_COLUMNS col
|
||||
LEFT JOIN
|
||||
DBA_COL_COMMENTS com
|
||||
ON col.OWNER = com.OWNER
|
||||
AND col.TABLE_NAME = com.TABLE_NAME
|
||||
AND col.COLUMN_NAME = com.COLUMN_NAME
|
||||
WHERE
|
||||
col.OWNER = '{conf.dbSchema}'
|
||||
"""
|
||||
sql2 = f" AND col.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "ck":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
name AS COLUMN_NAME,
|
||||
type AS DATA_TYPE,
|
||||
comment AS COLUMN_COMMENT
|
||||
FROM system.columns
|
||||
WHERE database = '{conf.database}'
|
||||
"""
|
||||
sql2 = f" AND table = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
db = DB.get_db(ds.type)
|
||||
if db.connect_type == ConnectType.sqlalchemy:
|
||||
with get_session(ds) as session:
|
||||
sql: str = ''
|
||||
if ds.type == "mysql":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
COLUMN_NAME,
|
||||
DATA_TYPE,
|
||||
COLUMN_COMMENT
|
||||
FROM
|
||||
INFORMATION_SCHEMA.COLUMNS
|
||||
WHERE
|
||||
TABLE_SCHEMA = '{conf.database}'
|
||||
"""
|
||||
sql2 = f" AND TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "sqlServer":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
COLUMN_NAME AS [COLUMN_NAME],
|
||||
DATA_TYPE AS [DATA_TYPE],
|
||||
ISNULL(EP.value, '') AS [COLUMN_COMMENT]
|
||||
FROM
|
||||
INFORMATION_SCHEMA.COLUMNS C
|
||||
LEFT JOIN
|
||||
sys.extended_properties EP
|
||||
ON EP.major_id = OBJECT_ID(C.TABLE_SCHEMA + '.' + C.TABLE_NAME)
|
||||
AND EP.minor_id = C.ORDINAL_POSITION
|
||||
AND EP.name = 'MS_Description'
|
||||
WHERE
|
||||
C.TABLE_SCHEMA = '{conf.dbSchema}'
|
||||
"""
|
||||
sql2 = f" AND C.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "pg" or ds.type == "excel":
|
||||
sql1 = """
|
||||
SELECT a.attname AS COLUMN_NAME,
|
||||
pg_catalog.format_type(a.atttypid, a.atttypmod) AS DATA_TYPE,
|
||||
col_description(c.oid, a.attnum) AS COLUMN_COMMENT
|
||||
FROM pg_catalog.pg_attribute a
|
||||
JOIN
|
||||
pg_catalog.pg_class c ON a.attrelid = c.oid
|
||||
JOIN
|
||||
pg_catalog.pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE n.nspname = current_schema()
|
||||
AND a.attnum > 0
|
||||
AND NOT a.attisdropped \
|
||||
"""
|
||||
sql2 = f" AND c.relname = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "oracle":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
col.COLUMN_NAME AS "COLUMN_NAME",
|
||||
(CASE
|
||||
WHEN col.DATA_TYPE IN ('VARCHAR2', 'CHAR', 'NVARCHAR2', 'NCHAR')
|
||||
THEN col.DATA_TYPE || '(' || col.DATA_LENGTH || ')'
|
||||
WHEN col.DATA_TYPE = 'NUMBER' AND col.DATA_PRECISION IS NOT NULL
|
||||
THEN col.DATA_TYPE || '(' || col.DATA_PRECISION ||
|
||||
CASE WHEN col.DATA_SCALE > 0 THEN ',' || col.DATA_SCALE END || ')'
|
||||
ELSE col.DATA_TYPE
|
||||
END) AS "DATA_TYPE",
|
||||
NVL(com.COMMENTS, '') AS "COLUMN_COMMENT"
|
||||
FROM
|
||||
DBA_TAB_COLUMNS col
|
||||
LEFT JOIN
|
||||
DBA_COL_COMMENTS com
|
||||
ON col.OWNER = com.OWNER
|
||||
AND col.TABLE_NAME = com.TABLE_NAME
|
||||
AND col.COLUMN_NAME = com.COLUMN_NAME
|
||||
WHERE
|
||||
col.OWNER = '{conf.dbSchema}'
|
||||
"""
|
||||
sql2 = f" AND col.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
elif ds.type == "ck":
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
name AS COLUMN_NAME,
|
||||
type AS DATA_TYPE,
|
||||
comment AS COLUMN_COMMENT
|
||||
FROM system.columns
|
||||
WHERE database = '{conf.database}'
|
||||
"""
|
||||
sql2 = f" AND table = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
sql = sql1 + sql2
|
||||
|
||||
with session.execute(text(sql)) as result:
|
||||
res = result.fetchall()
|
||||
res_list = [ColumnSchema(*item) for item in res]
|
||||
return res_list
|
||||
|
||||
|
||||
def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str):
|
||||
with get_session(ds) as session:
|
||||
with session.execute(text(sql)) as result:
|
||||
try:
|
||||
columns = [item.lower() for item in result.keys()._keys]
|
||||
with session.execute(text(sql)) as result:
|
||||
res = result.fetchall()
|
||||
result_list = [
|
||||
{str(columns[i]): float(value) if isinstance(value, Decimal) else value for i, value in
|
||||
enumerate(tuple_item)}
|
||||
for tuple_item in res
|
||||
]
|
||||
return {"fields": columns, "data": result_list,
|
||||
"sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))}
|
||||
except Exception as ex:
|
||||
raise ex
|
||||
res_list = [ColumnSchema(*item) for item in res]
|
||||
return res_list
|
||||
else:
|
||||
if ds.type == 'dm':
|
||||
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
|
||||
port=conf.port) as conn, conn.cursor() as cursor:
|
||||
sql1 = f"""
|
||||
SELECT
|
||||
c.COLUMN_NAME AS "COLUMN_NAME",
|
||||
c.DATA_TYPE AS "DATA_TYPE",
|
||||
COALESCE(com.COMMENTS, '') AS "COMMENTS"
|
||||
FROM
|
||||
ALL_TAB_COLS c
|
||||
LEFT JOIN
|
||||
ALL_COL_COMMENTS com
|
||||
ON c.OWNER = com.OWNER
|
||||
AND c.TABLE_NAME = com.TABLE_NAME
|
||||
AND c.COLUMN_NAME = com.COLUMN_NAME
|
||||
WHERE
|
||||
c.OWNER = '{conf.dbSchema}'
|
||||
"""
|
||||
sql2 = f" AND c.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
|
||||
cursor.execute(sql1 + sql2)
|
||||
res = cursor.fetchall()
|
||||
res_list = [ColumnSchema(*item) for item in res]
|
||||
return res_list
|
||||
|
||||
|
||||
def exec_sql(ds: CoreDatasource | AssistantOutDsSchema, sql: str, origin_column=False):
|
||||
db = DB.get_db(ds.type)
|
||||
if db.connect_type == ConnectType.sqlalchemy:
|
||||
with get_session(ds) as session:
|
||||
with session.execute(text(sql)) as result:
|
||||
try:
|
||||
columns = result.keys()._keys if origin_column else [item.lower() for item in result.keys()._keys]
|
||||
res = result.fetchall()
|
||||
result_list = [
|
||||
{str(columns[i]): float(value) if isinstance(value, Decimal) else value for i, value in
|
||||
enumerate(tuple_item)}
|
||||
for tuple_item in res
|
||||
]
|
||||
return {"fields": columns, "data": result_list,
|
||||
"sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))}
|
||||
except Exception as ex:
|
||||
raise ex
|
||||
else:
|
||||
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration)))
|
||||
if ds.type == 'dm':
|
||||
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
|
||||
port=conf.port) as conn, conn.cursor() as cursor:
|
||||
try:
|
||||
cursor.execute(sql)
|
||||
res = cursor.fetchall()
|
||||
columns = [field[0] for field in cursor.description] if origin_column else [field[0].lower() for
|
||||
field in
|
||||
cursor.description]
|
||||
result_list = [
|
||||
{str(columns[i]): float(value) if isinstance(value, Decimal) else value for i, value in
|
||||
enumerate(tuple_item)}
|
||||
for tuple_item in res
|
||||
]
|
||||
return {"fields": columns, "data": result_list,
|
||||
"sql": bytes.decode(base64.b64encode(bytes(sql, 'utf-8')))}
|
||||
except Exception as ex:
|
||||
raise ex
|
||||
|
||||
@@ -10,5 +10,6 @@ def db_type_relation() -> Dict:
|
||||
"pg": "PostgreSQL",
|
||||
"excel": "Excel/CSV",
|
||||
"oracle": "Oracle",
|
||||
"ck": "ClickHouse"
|
||||
"ck": "ClickHouse",
|
||||
"dm": "达梦"
|
||||
}
|
||||
|
||||
@@ -44,6 +44,7 @@ dependencies = [
|
||||
"python-calamine>=0.4.0",
|
||||
"xlrd>=2.0.2",
|
||||
"clickhouse-sqlalchemy>=0.3.2",
|
||||
"dmpython>=2.5.22",
|
||||
]
|
||||
[[tool.uv.index]]
|
||||
name = "default"
|
||||
|
||||
@@ -27,7 +27,7 @@ template:
|
||||
- SQL查询的字段若是函数字段,如 COUNT(),CAST() 等,必须加上别名
|
||||
- 计算占比,百分比类型字段,保留两位小数,以%结尾。
|
||||
- 生成SQL时,必须避免关键字冲突。
|
||||
- 如数据库引擎是 PostgreSQL、Oracle、ClickHouse,则在schema、表名、字段名、别名外层加双引号;
|
||||
- 如数据库引擎是 PostgreSQL、Oracle、ClickHouse、达梦(DM),则在schema、表名、字段名、别名外层加双引号;
|
||||
- 如数据库引擎是 MySQL,则在表名、字段名、别名外层加反引号;
|
||||
- 如数据库引擎是 Microsoft SQL Server,则在schema、表名、字段名、别名外层加方括号。
|
||||
- 以PostgreSQL为例,查询Schema为TEST表TABLE下所有数据,则生成的SQL为:
|
||||
@@ -208,7 +208,7 @@ template:
|
||||
- 如果存在冗余的过滤条件则进行去重后再生成新SQL。
|
||||
- 给过滤条件中的字段前加上表别名(如果没有表别名则加表名),如:table.field。
|
||||
- 生成SQL时,必须避免关键字冲突:
|
||||
- 如数据库引擎是 PostgreSQL、Oracle,则在schema、表名、字段名、别名外层加双引号;
|
||||
- 如数据库引擎是 PostgreSQL、Oracle、ClickHouse、达梦(DM),则在schema、表名、字段名、别名外层加双引号;
|
||||
- 如数据库引擎是 MySQL,则在表名、字段名、别名外层加反引号;
|
||||
- 如数据库引擎是 Microsoft SQL Server,则在schema、表名、字段名、别名外层加方括号。
|
||||
- 生成的SQL使用JSON格式返回:
|
||||
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 1.3 KiB |
@@ -616,7 +616,7 @@ defineExpose({
|
||||
show-password
|
||||
/>
|
||||
</el-form-item>
|
||||
<el-form-item :label="t('ds.form.database')" prop="database">
|
||||
<el-form-item v-if="form.type !== 'dm'" :label="t('ds.form.database')" prop="database">
|
||||
<el-input
|
||||
v-model="form.database"
|
||||
clearable
|
||||
|
||||
@@ -4,6 +4,7 @@ import oracle from '@/assets/datasource/icon_oracle.png'
|
||||
import pg from '@/assets/datasource/icon_PostgreSQL.png'
|
||||
import sqlServer from '@/assets/datasource/icon_SQL_Server.png'
|
||||
import ck from '@/assets/datasource/icon_ck.png'
|
||||
import dm from '@/assets/datasource/icon_dm.png'
|
||||
import { i18n } from '@/i18n'
|
||||
|
||||
const t = i18n.global.t
|
||||
@@ -14,6 +15,7 @@ export const dsType = [
|
||||
{ label: 'PostgreSQL', value: 'pg' },
|
||||
{ label: 'SQL Server', value: 'sqlServer' },
|
||||
{ label: 'ClickHouse', value: 'ck' },
|
||||
{ label: '达梦', value: 'dm' },
|
||||
]
|
||||
|
||||
export const dsTypeWithImg = [
|
||||
@@ -23,6 +25,7 @@ export const dsTypeWithImg = [
|
||||
{ name: 'PostgreSQL', type: 'pg', img: pg },
|
||||
{ name: 'SQL Server', type: 'sqlServer', img: sqlServer },
|
||||
{ name: 'ClickHouse', type: 'ck', img: ck },
|
||||
{ name: '达梦', type: 'dm', img: dm },
|
||||
]
|
||||
|
||||
export const haveSchema = ['sqlServer', 'pg', 'oracle']
|
||||
export const haveSchema = ['sqlServer', 'pg', 'oracle', 'dm']
|
||||
|
||||
Reference in New Issue
Block a user