Skip to content

Commit db2fdfc

Browse files
committed
refactor: code optimization
1 parent 418ad4d commit db2fdfc

File tree

3 files changed

+227
-209
lines changed

3 files changed

+227
-209
lines changed

backend/apps/datasource/crud/datasource.py

Lines changed: 3 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -1,23 +1,19 @@
11
import datetime
22
import json
3-
import platform
43
from typing import List, Optional
54

6-
if platform.system() != "Darwin":
7-
import dmPython
85
from fastapi import HTTPException
96
from sqlalchemy import and_, text
107
from sqlmodel import select
118

129
from apps.datasource.crud.permission import get_column_permission_fields, get_row_permission_filters, is_normal_user
1310
from apps.datasource.utils.utils import aes_decrypt
14-
from apps.db.constant import ConnectType
1511
from apps.db.constant import DB
16-
from apps.db.db import get_engine, get_tables, get_fields, exec_sql
12+
from apps.db.db import get_tables, get_fields, exec_sql, check_connection
1713
from apps.db.engine import get_engine_config, get_engine_conn
1814
from apps.db.type import db_type_relation
1915
from common.core.deps import SessionDep, CurrentUser, Trans
20-
from common.utils.utils import SQLBotLogUtil, deepcopy_ignore_extra
16+
from common.utils.utils import deepcopy_ignore_extra
2117
from .table import get_tables_by_ds_id
2218
from ..crud.field import delete_field_by_ds_id, update_field
2319
from ..crud.table import delete_table_by_ds_id, update_table
@@ -49,32 +45,7 @@ def check_status_by_id(session: SessionDep, trans: Trans, ds_id: int, is_raise:
4945

5046

5147
def check_status(session: SessionDep, trans: Trans, ds: CoreDatasource, is_raise: bool = False):
52-
db = DB.get_db(ds.type)
53-
if db.connect_type == ConnectType.sqlalchemy:
54-
conn = get_engine(ds, 10)
55-
try:
56-
with conn.connect() as connection:
57-
SQLBotLogUtil.info("success")
58-
return True
59-
except Exception as e:
60-
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
61-
if is_raise:
62-
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
63-
return False
64-
else:
65-
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration)))
66-
if ds.type == 'dm':
67-
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
68-
port=conf.port) as conn, conn.cursor() as cursor:
69-
try:
70-
cursor.execute('select 1', timeout=10).fetchall()
71-
SQLBotLogUtil.info("success")
72-
return True
73-
except Exception as e:
74-
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
75-
if is_raise:
76-
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
77-
return False
48+
return check_connection(trans, ds, is_raise)
7849

7950

8051
def check_name(session: SessionDep, trans: Trans, user: CurrentUser, ds: CoreDatasource):

backend/apps/db/db.py

Lines changed: 38 additions & 177 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,8 @@
44
import urllib.parse
55
from decimal import Decimal
66

7+
from apps.db.db_sql import get_table_sql, get_field_sql
8+
79
if platform.system() != "Darwin":
810
import dmPython
911
from sqlalchemy import create_engine, text, Engine
@@ -15,6 +17,9 @@
1517
from apps.db.engine import get_engine_config
1618
from apps.system.crud.assistant import get_ds_engine
1719
from apps.system.schemas.system_schema import AssistantOutDsSchema
20+
from common.core.deps import Trans
21+
from common.utils.utils import SQLBotLogUtil
22+
from fastapi import HTTPException
1823

1924

2025
def get_uri(ds: CoreDatasource) -> str:
@@ -93,6 +98,35 @@ def get_session(ds: CoreDatasource | AssistantOutDsSchema):
9398
return session
9499

95100

101+
def check_connection(trans: Trans, ds: CoreDatasource, is_raise: bool = False):
102+
db = DB.get_db(ds.type)
103+
if db.connect_type == ConnectType.sqlalchemy:
104+
conn = get_engine(ds, 10)
105+
try:
106+
with conn.connect() as connection:
107+
SQLBotLogUtil.info("success")
108+
return True
109+
except Exception as e:
110+
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
111+
if is_raise:
112+
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
113+
return False
114+
else:
115+
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration)))
116+
if ds.type == 'dm':
117+
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
118+
port=conf.port) as conn, conn.cursor() as cursor:
119+
try:
120+
cursor.execute('select 1', timeout=10).fetchall()
121+
SQLBotLogUtil.info("success")
122+
return True
123+
except Exception as e:
124+
SQLBotLogUtil.error(f"Datasource {ds.id} connection failed: {e}")
125+
if is_raise:
126+
raise HTTPException(status_code=500, detail=trans('i18n_ds_invalid') + f': {e.args}')
127+
return False
128+
129+
96130
def get_schema(ds: CoreDatasource):
97131
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config()
98132
db = DB.get_db(ds.type)
@@ -122,78 +156,9 @@ def get_schema(ds: CoreDatasource):
122156
def get_tables(ds: CoreDatasource):
123157
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config()
124158
db = DB.get_db(ds.type)
159+
sql = get_table_sql(ds, conf)
125160
if db.connect_type == ConnectType.sqlalchemy:
126161
with get_session(ds) as session:
127-
sql: str = ''
128-
if ds.type == "mysql":
129-
sql = f"""
130-
SELECT
131-
TABLE_NAME,
132-
TABLE_COMMENT
133-
FROM
134-
information_schema.TABLES
135-
WHERE
136-
TABLE_SCHEMA = '{conf.database}'
137-
"""
138-
elif ds.type == "sqlServer":
139-
sql = f"""
140-
SELECT
141-
TABLE_NAME AS [TABLE_NAME],
142-
ISNULL(ep.value, '') AS [TABLE_COMMENT]
143-
FROM
144-
INFORMATION_SCHEMA.TABLES t
145-
LEFT JOIN
146-
sys.extended_properties ep
147-
ON ep.major_id = OBJECT_ID(t.TABLE_SCHEMA + '.' + t.TABLE_NAME)
148-
AND ep.minor_id = 0
149-
AND ep.name = 'MS_Description'
150-
WHERE
151-
t.TABLE_TYPE IN ('BASE TABLE', 'VIEW')
152-
AND t.TABLE_SCHEMA = '{conf.dbSchema}'
153-
"""
154-
elif ds.type == "pg" or ds.type == "excel":
155-
sql = """
156-
SELECT c.relname AS TABLE_NAME,
157-
COALESCE(d.description, obj_description(c.oid)) AS TABLE_COMMENT
158-
FROM pg_class c
159-
LEFT JOIN
160-
pg_namespace n ON n.oid = c.relnamespace
161-
LEFT JOIN
162-
pg_description d ON d.objoid = c.oid AND d.objsubid = 0
163-
WHERE n.nspname = current_schema()
164-
AND c.relkind IN ('r', 'v', 'p', 'm')
165-
AND c.relname NOT LIKE 'pg_%'
166-
AND c.relname NOT LIKE 'sql_%'
167-
ORDER BY c.relname \
168-
"""
169-
elif ds.type == "oracle":
170-
sql = f"""
171-
SELECT
172-
t.TABLE_NAME AS "TABLE_NAME",
173-
NVL(c.COMMENTS, '') AS "TABLE_COMMENT"
174-
FROM (
175-
SELECT TABLE_NAME, 'TABLE' AS OBJECT_TYPE
176-
FROM DBA_TABLES
177-
WHERE OWNER = '{conf.dbSchema}'
178-
UNION ALL
179-
SELECT VIEW_NAME AS TABLE_NAME, 'VIEW' AS OBJECT_TYPE
180-
FROM DBA_VIEWS
181-
WHERE OWNER = '{conf.dbSchema}'
182-
) t
183-
LEFT JOIN DBA_TAB_COMMENTS c
184-
ON t.TABLE_NAME = c.TABLE_NAME
185-
AND c.TABLE_TYPE = t.OBJECT_TYPE
186-
AND c.OWNER = '{conf.dbSchema}'
187-
ORDER BY t.TABLE_NAME
188-
"""
189-
elif ds.type == "ck":
190-
sql = f"""
191-
SELECT name, comment
192-
FROM system.tables
193-
WHERE database = '{conf.database}'
194-
AND engine NOT IN ('Dictionary')
195-
ORDER BY name
196-
"""
197162
with session.execute(text(sql)) as result:
198163
res = result.fetchall()
199164
res_list = [TableSchema(*item) for item in res]
@@ -202,11 +167,7 @@ def get_tables(ds: CoreDatasource):
202167
if ds.type == 'dm':
203168
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
204169
port=conf.port) as conn, conn.cursor() as cursor:
205-
cursor.execute(f"""select table_name, comments
206-
from all_tab_comments
207-
where owner='{conf.dbSchema}'
208-
AND (table_type = 'TABLE' or table_type = 'VIEW')
209-
""", timeout=conf.timeout)
170+
cursor.execute(sql, timeout=conf.timeout)
210171
res = cursor.fetchall()
211172
res_list = [TableSchema(*item) for item in res]
212173
return res_list
@@ -215,93 +176,9 @@ def get_tables(ds: CoreDatasource):
215176
def get_fields(ds: CoreDatasource, table_name: str = None):
216177
conf = DatasourceConf(**json.loads(aes_decrypt(ds.configuration))) if ds.type != "excel" else get_engine_config()
217178
db = DB.get_db(ds.type)
179+
sql = get_field_sql(ds, conf, table_name)
218180
if db.connect_type == ConnectType.sqlalchemy:
219181
with get_session(ds) as session:
220-
sql: str = ''
221-
if ds.type == "mysql":
222-
sql1 = f"""
223-
SELECT
224-
COLUMN_NAME,
225-
DATA_TYPE,
226-
COLUMN_COMMENT
227-
FROM
228-
INFORMATION_SCHEMA.COLUMNS
229-
WHERE
230-
TABLE_SCHEMA = '{conf.database}'
231-
"""
232-
sql2 = f" AND TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
233-
sql = sql1 + sql2
234-
elif ds.type == "sqlServer":
235-
sql1 = f"""
236-
SELECT
237-
COLUMN_NAME AS [COLUMN_NAME],
238-
DATA_TYPE AS [DATA_TYPE],
239-
ISNULL(EP.value, '') AS [COLUMN_COMMENT]
240-
FROM
241-
INFORMATION_SCHEMA.COLUMNS C
242-
LEFT JOIN
243-
sys.extended_properties EP
244-
ON EP.major_id = OBJECT_ID(C.TABLE_SCHEMA + '.' + C.TABLE_NAME)
245-
AND EP.minor_id = C.ORDINAL_POSITION
246-
AND EP.name = 'MS_Description'
247-
WHERE
248-
C.TABLE_SCHEMA = '{conf.dbSchema}'
249-
"""
250-
sql2 = f" AND C.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
251-
sql = sql1 + sql2
252-
elif ds.type == "pg" or ds.type == "excel":
253-
sql1 = """
254-
SELECT a.attname AS COLUMN_NAME,
255-
pg_catalog.format_type(a.atttypid, a.atttypmod) AS DATA_TYPE,
256-
col_description(c.oid, a.attnum) AS COLUMN_COMMENT
257-
FROM pg_catalog.pg_attribute a
258-
JOIN
259-
pg_catalog.pg_class c ON a.attrelid = c.oid
260-
JOIN
261-
pg_catalog.pg_namespace n ON n.oid = c.relnamespace
262-
WHERE n.nspname = current_schema()
263-
AND a.attnum > 0
264-
AND NOT a.attisdropped \
265-
"""
266-
sql2 = f" AND c.relname = '{table_name}'" if table_name is not None and table_name != "" else ""
267-
sql = sql1 + sql2
268-
elif ds.type == "oracle":
269-
sql1 = f"""
270-
SELECT
271-
col.COLUMN_NAME AS "COLUMN_NAME",
272-
(CASE
273-
WHEN col.DATA_TYPE IN ('VARCHAR2', 'CHAR', 'NVARCHAR2', 'NCHAR')
274-
THEN col.DATA_TYPE || '(' || col.DATA_LENGTH || ')'
275-
WHEN col.DATA_TYPE = 'NUMBER' AND col.DATA_PRECISION IS NOT NULL
276-
THEN col.DATA_TYPE || '(' || col.DATA_PRECISION ||
277-
CASE WHEN col.DATA_SCALE > 0 THEN ',' || col.DATA_SCALE END || ')'
278-
ELSE col.DATA_TYPE
279-
END) AS "DATA_TYPE",
280-
NVL(com.COMMENTS, '') AS "COLUMN_COMMENT"
281-
FROM
282-
DBA_TAB_COLUMNS col
283-
LEFT JOIN
284-
DBA_COL_COMMENTS com
285-
ON col.OWNER = com.OWNER
286-
AND col.TABLE_NAME = com.TABLE_NAME
287-
AND col.COLUMN_NAME = com.COLUMN_NAME
288-
WHERE
289-
col.OWNER = '{conf.dbSchema}'
290-
"""
291-
sql2 = f" AND col.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
292-
sql = sql1 + sql2
293-
elif ds.type == "ck":
294-
sql1 = f"""
295-
SELECT
296-
name AS COLUMN_NAME,
297-
type AS DATA_TYPE,
298-
comment AS COLUMN_COMMENT
299-
FROM system.columns
300-
WHERE database = '{conf.database}'
301-
"""
302-
sql2 = f" AND table = '{table_name}'" if table_name is not None and table_name != "" else ""
303-
sql = sql1 + sql2
304-
305182
with session.execute(text(sql)) as result:
306183
res = result.fetchall()
307184
res_list = [ColumnSchema(*item) for item in res]
@@ -310,23 +187,7 @@ def get_fields(ds: CoreDatasource, table_name: str = None):
310187
if ds.type == 'dm':
311188
with dmPython.connect(user=conf.username, password=conf.password, server=conf.host,
312189
port=conf.port) as conn, conn.cursor() as cursor:
313-
sql1 = f"""
314-
SELECT
315-
c.COLUMN_NAME AS "COLUMN_NAME",
316-
c.DATA_TYPE AS "DATA_TYPE",
317-
COALESCE(com.COMMENTS, '') AS "COMMENTS"
318-
FROM
319-
ALL_TAB_COLS c
320-
LEFT JOIN
321-
ALL_COL_COMMENTS com
322-
ON c.OWNER = com.OWNER
323-
AND c.TABLE_NAME = com.TABLE_NAME
324-
AND c.COLUMN_NAME = com.COLUMN_NAME
325-
WHERE
326-
c.OWNER = '{conf.dbSchema}'
327-
"""
328-
sql2 = f" AND c.TABLE_NAME = '{table_name}'" if table_name is not None and table_name != "" else ""
329-
cursor.execute(sql1 + sql2, timeout=conf.timeout)
190+
cursor.execute(sql, timeout=conf.timeout)
330191
res = cursor.fetchall()
331192
res_list = [ColumnSchema(*item) for item in res]
332193
return res_list

0 commit comments

Comments
 (0)