From a881d4700d4c3c610465384c4f9681c2405e581a Mon Sep 17 00:00:00 2001 From: junjun Date: Thu, 15 May 2025 15:51:57 +0800 Subject: [PATCH] feat: sync tables and fields --- .../0a6f11be9be4_005_table_and_field.py | 49 ++++++++++ backend/apps/datasource/api/datasource.py | 4 +- backend/apps/datasource/crud/datasource.py | 89 +++++++++++++++++-- backend/apps/datasource/crud/field.py | 7 ++ backend/apps/datasource/crud/table.py | 7 ++ backend/apps/datasource/models/datasource.py | 45 +++++++--- backend/common/utils/utils.py | 10 +++ frontend/src/views/ds/form.vue | 1 + 8 files changed, 191 insertions(+), 21 deletions(-) create mode 100644 backend/alembic/versions/0a6f11be9be4_005_table_and_field.py create mode 100644 backend/apps/datasource/crud/field.py create mode 100644 backend/apps/datasource/crud/table.py diff --git a/backend/alembic/versions/0a6f11be9be4_005_table_and_field.py b/backend/alembic/versions/0a6f11be9be4_005_table_and_field.py new file mode 100644 index 00000000..0d65f009 --- /dev/null +++ b/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 ### diff --git a/backend/apps/datasource/api/datasource.py b/backend/apps/datasource/api/datasource.py index 924ed850..9d821b24 100644 --- a/backend/apps/datasource/api/datasource.py +++ b/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) diff --git a/backend/apps/datasource/crud/datasource.py b/backend/apps/datasource/crud/datasource.py index df543d64..20f2af62 100644 --- a/backend/apps/datasource/crud/datasource.py +++ b/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) diff --git a/backend/apps/datasource/crud/field.py b/backend/apps/datasource/crud/field.py new file mode 100644 index 00000000..84c4fafc --- /dev/null +++ b/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() diff --git a/backend/apps/datasource/crud/table.py b/backend/apps/datasource/crud/table.py new file mode 100644 index 00000000..7a84a478 --- /dev/null +++ b/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() diff --git a/backend/apps/datasource/models/datasource.py b/backend/apps/datasource/models/datasource.py index e0cd1f58..e8af9f3b 100644 --- a/backend/apps/datasource/models/datasource.py +++ b/backend/apps/datasource/models/datasource.py @@ -2,7 +2,7 @@ from sqlmodel import SQLModel, Field 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(BaseModel): - checked: str = True - field_name: str = '' - field_type: str = '' - field_comment: str = '' - custom_comment: str = '' +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)) + + +# 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): diff --git a/backend/common/utils/utils.py b/backend/common/utils/utils.py index b0099573..d168decd 100644 --- a/backend/common/utils/utils.py +++ b/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 \ No newline at end of file diff --git a/frontend/src/views/ds/form.vue b/frontend/src/views/ds/form.vue index 2d3caae7..eecca175 100644 --- a/frontend/src/views/ds/form.vue +++ b/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()