klinsmaya/SQLBot
1
0
Fork 0
mirror of https://gh-proxy.org/https://github.com/klinsmaya/SQLBot.git synced 2026-11-08 09:17:06 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: DataSource check

Browse Source
This commit is contained in:
junjun
2025-06-17 14:34:52 +08:00
parent eb80b515f0
commit 6c3c612ada
1 changed files with 14 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/datasource/crud/datasource.py
+14
View File
@@ -41,9 +41,22 @@ def check_status(session: SessionDep, ds: CoreDatasource):
return False
def check_name(session: SessionDep, ds: CoreDatasource):
if ds.id is not None:
ds_list = session.query(CoreDatasource).filter(
and_(CoreDatasource.name == ds.name, CoreDatasource.id != ds.id)).all()
if ds_list is not None and len(ds_list) > 0:
raise 'Name exist'
else:
ds_list = session.query(CoreDatasource).filter(CoreDatasource.name == ds.name).all()
if ds_list is not None and len(ds_list) > 0:
raise 'Name exist'
def create_ds(session: SessionDep, user: CurrentUser, create_ds: CreateDatasource):
ds = CoreDatasource()
deepcopy_ignore_extra(create_ds, ds)
check_name(session, ds)
ds.create_time = datetime.datetime.now()
# status = check_status(session, ds)
ds.create_by = user.id
@@ -68,6 +81,7 @@ def chooseTables(session: SessionDep, id: int, tables: List[CoreTable]):
def update_ds(session: SessionDep, ds: CoreDatasource):
ds.id = int(ds.id)
check_name(session, ds)
status = check_status(session, ds)
ds.status = "Success" if status is True else "Fail"
record = session.exec(select(CoreDatasource).where(CoreDatasource.id == ds.id)).first()