feat: sync tables and fields

Author: XiaJunjie2020

backend/alembic/versions/0a6f11be9be4_005_table_and_field.py

@@ -0,0 +1,49 @@
+"""005_table_and_field
+
+Revision ID: 0a6f11be9be4
+Revises: 8fe654655905
+Create Date: 2025-05-15 10:20:25.686576
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = '0a6f11be9be4'
+down_revision = '8fe654655905'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('core_field',
+    sa.Column('id', sa.Integer(), sa.Identity(always=True), nullable=False),
+    sa.Column('ds_id', sa.BigInteger(), nullable=True),
+    sa.Column('table_id', sa.BigInteger(), nullable=True),
+    sa.Column('checked', sa.Boolean(), nullable=False),
+    sa.Column('field_name', sa.Text(), nullable=True),
+    sa.Column('field_type', sqlmodel.sql.sqltypes.AutoString(length=128), nullable=True),
+    sa.Column('field_comment', sa.Text(), nullable=True),
+    sa.Column('custom_comment', sa.Text(), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('core_table',
+    sa.Column('id', sa.Integer(), sa.Identity(always=True), nullable=False),
+    sa.Column('ds_id', sa.BigInteger(), nullable=True),
+    sa.Column('checked', sa.Boolean(), nullable=False),
+    sa.Column('table_name', sa.Text(), nullable=True),
+    sa.Column('table_comment', sa.Text(), nullable=True),
+    sa.Column('custom_comment', sa.Text(), nullable=True),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('core_table')
+    op.drop_table('core_field')
+    # ### end Alembic commands ###

backend/apps/datasource/api/datasource.py

@@ -1,7 +1,7 @@
 from fastapi import APIRouter
 from ..crud.datasource import get_datasource_list, check_status, create_ds, update_ds, delete_ds, getTables, getFields, execSql
 from common.core.deps import SessionDep
-from ..models.datasource import CoreDatasource
+from ..models.datasource import CoreDatasource, CreateDatasource
 
 router = APIRouter(tags=["datasource"], prefix="/datasource")
 
@@ -17,7 +17,7 @@ async def check(session: SessionDep, ds: CoreDatasource):
 
 
 @router.post("/add", response_model=CoreDatasource)
-async def add(session: SessionDep, ds: CoreDatasource):
+async def add(session: SessionDep, ds: CreateDatasource):
     return create_ds(session, ds)
 
 

backend/apps/datasource/crud/datasource.py

@@ -1,9 +1,13 @@
 from sqlmodel import select
-from ..models.datasource import CoreDatasource, DatasourceConf
+from ..models.datasource import CoreDatasource, CreateDatasource, CoreTable, CoreField, ColumnSchema
 import datetime
 from common.core.deps import SessionDep
 from apps.db.db import get_session, get_tables, get_fields, exec_sql
-from sqlalchemy import text
+from sqlalchemy import text, and_
+from common.utils.utils import deepcopy_ignore_extra
+from typing import List
+from ..crud.table import delete_table_by_ds_id
+from ..crud.field import delete_field_by_ds_id
 
 
 def get_datasource_list(session: SessionDep) -> CoreDatasource:
@@ -26,21 +30,30 @@ def check_status(session: SessionDep, ds: CoreDatasource):
     return False
 
 
-def create_ds(session: SessionDep, ds: CoreDatasource):
+def create_ds(session: SessionDep, create_ds: CreateDatasource):
+    ds = CoreDatasource()
+    deepcopy_ignore_extra(create_ds, ds)
     ds.create_time = datetime.datetime.now()
     status = check_status(session, ds)
     ds.status = "Success" if status is True else "Fail"
     record = CoreDatasource(**ds.model_dump())
-    # get tables and fields
-    if status:
-        pass
     session.add(record)
+    session.flush()
+    session.refresh(record)
+    ds.id = record.id
     session.commit()
+
+    # save tables and fields
+    if status:
+        sync_table(session, ds, create_ds.tables)
+
     return ds
 
 
 def update_ds(session: SessionDep, ds: CoreDatasource):
     ds.id = int(ds.id)
+    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()
     update_data = ds.model_dump(exclude_unset=True)
     for field, value in update_data.items():
@@ -54,6 +67,8 @@ def delete_ds(session: SessionDep, id: int):
     term = session.exec(select(CoreDatasource).where(CoreDatasource.id == id)).first()
     session.delete(term)
     session.commit()
+    delete_table_by_ds_id(session, id)
+    delete_field_by_ds_id(session, id)
     return {
         "message": f"Datasource with ID {id} deleted successfully."
     }
@@ -74,3 +89,65 @@ def getFields(session: SessionDep, id: int, 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)
+
+
+def sync_table(session: SessionDep, ds: CoreDatasource, tables: List[CoreTable]):
+    id_list = []
+    for item in tables:
+        statement = select(CoreTable).where(and_(CoreTable.ds_id == ds.id, CoreTable.table_name == item.table_name))
+        record = session.exec(statement).first()
+        # update exist table, only update table_comment
+        if record is not None:
+            item.id = record.id
+            id_list.append(record.id)
+
+            record.table_comment = item.table_comment
+            session.add(record)
+            session.commit()
+        else:
+            # save new table
+            table = CoreTable(ds_id=ds.id, checked=True, table_name=item.table_name, table_comment=item.table_comment,
+                              custom_comment=item.table_comment)
+            session.add(table)
+            session.flush()
+            session.refresh(table)
+            item.id = table.id
+            id_list.append(table.id)
+            session.commit()
+
+        # sync field
+        fields = getFields(session, ds.id, item.table_name)
+        sync_fields(session, ds, item, fields)
+
+    if len(id_list) > 0:
+        session.query(CoreTable).filter(and_(CoreTable.ds_id == ds.id, CoreTable.id.not_in(id_list))).delete(
+            synchronize_session=False)
+
+
+def sync_fields(session: SessionDep, ds: CoreDatasource, table: CoreTable, fields: List[ColumnSchema]):
+    id_list = []
+    for item in fields:
+        statement = select(CoreField).where(
+            and_(CoreField.table_id == table.id, CoreField.field_name == item.fieldName))
+        record = session.exec(statement).first()
+        if record is not None:
+            item.id = record.id
+            id_list.append(record.id)
+
+            record.field_comment = item.fieldComment
+            session.add(record)
+            session.commit()
+        else:
+            field = CoreField(ds_id=ds.id, table_id=table.id, checked=True, field_name=item.fieldName,
+                              field_type=item.fieldType, field_comment=item.fieldComment,
+                              custom_comment=item.fieldComment)
+            session.add(field)
+            session.flush()
+            session.refresh(field)
+            item.id = field.id
+            id_list.append(field.id)
+            session.commit()
+
+    if len(id_list) > 0:
+        session.query(CoreField).filter(and_(CoreField.table_id == table.id, CoreField.id.not_in(id_list))).delete(
+            synchronize_session=False)

backend/apps/datasource/crud/field.py

@@ -0,0 +1,7 @@
+from common.core.deps import SessionDep
+from ..models.datasource import CoreDatasource, CreateDatasource, CoreTable, CoreField, ColumnSchema
+
+
+def delete_field_by_ds_id(session: SessionDep, id: int):
+    session.query(CoreField).filter(CoreField.ds_id == id).delete(synchronize_session=False)
+    session.commit()

backend/apps/datasource/crud/table.py

@@ -0,0 +1,7 @@
+from common.core.deps import SessionDep
+from ..models.datasource import CoreDatasource, CreateDatasource, CoreTable, CoreField, ColumnSchema
+
+
+def delete_table_by_ds_id(session: SessionDep, id: int):
+    session.query(CoreTable).filter(CoreTable.ds_id == id).delete(synchronize_session=False)
+    session.commit()

backend/apps/datasource/models/datasource.py

@@ -2,7 +2,7 @@
 from sqlalchemy import Column, Text, BigInteger, DateTime, Integer, Identity
 from datetime import datetime
 from pydantic import BaseModel
-from typing import List
+from typing import List, Optional
 
 
 class CoreDatasource(SQLModel, table=True):
@@ -17,20 +17,39 @@ class CoreDatasource(SQLModel, table=True):
     status: str = Field(max_length=64, nullable=True)
 
 
-class CoreTable(BaseModel):
-    checked: str = True
-    table_name: str = ''
-    table_comment: str = ''
-    custom_comment: str = ''
-    table_fields: List = []
+class CoreTable(SQLModel, table=True):
+    __tablename__ = "core_table"
+    id: int = Field(sa_column=Column(Integer, Identity(always=True), nullable=False, primary_key=True))
+    ds_id:int = Field(sa_column=Column(BigInteger()))
+    checked: bool = Field(default=True)
+    table_name: str = Field(sa_column=Column(Text))
+    table_comment: str = Field(sa_column=Column(Text))
+    custom_comment: str = Field(sa_column=Column(Text))
+
+
+class CoreField(SQLModel, table=True):
+    __tablename__ = "core_field"
+    id: int = Field(sa_column=Column(Integer, Identity(always=True), nullable=False, primary_key=True))
+    ds_id: int = Field(sa_column=Column(BigInteger()))
+    table_id: int = Field(sa_column=Column(BigInteger()))
+    checked: bool = Field(default=True)
+    field_name: str = Field(sa_column=Column(Text))
+    field_type: str = Field(max_length=128, nullable=True)
+    field_comment: str = Field(sa_column=Column(Text))
+    custom_comment: str = Field(sa_column=Column(Text))
 
 
-class CoreField(BaseModel):
-    checked: str = True
-    field_name: str = ''
-    field_type: str = ''
-    field_comment: str = ''
-    custom_comment: str = ''
+# datasource create obj
+class CreateDatasource(BaseModel):
+    id: int = None
+    name: str = ''
+    description: str = ''
+    type: str = ''
+    configuration: str = ''
+    create_time: Optional[datetime] = None
+    create_by: int = 0
+    status: str = ''
+    tables: List[CoreTable] = []
 
 
 class DatasourceConf(BaseModel):

backend/common/utils/utils.py

@@ -37,3 +37,13 @@ def verify_password_reset_token(token: str) -> str | None:
         return str(decoded_token["sub"])
     except InvalidTokenError:
         return None
+
+
+def deepcopy_ignore_extra(src, dest):
+    import copy
+    for attr in vars(src):
+        if hasattr(dest, attr):
+            src_value = getattr(src, attr)
+            dest_value = copy.deepcopy(src_value)  # deep copy
+            setattr(dest, attr, dest_value)
+    return dest

frontend/src/views/ds/form.vue

@@ -137,6 +137,7 @@ const save = async(formEl: FormInstance | undefined) => {
           emit('refresh')
         })
       } else {
+        form.value.tables = [{"table_name":"core_dataset_table"},{"table_name":"pre_user"}]
         datasourceApi.add(form.value).then((res: any) => {
           console.log(res)
           close()