12 changed files with 558 additions and 130 deletions
@ -0,0 +1,155 @@ |
|||
from sqlalchemy import delete, select, update, desc,or_,not_,insert,and_,func |
|||
from sqlalchemy.ext.asyncio import AsyncSession |
|||
from module_admin.entity.do.dataast_do import DataAssetInfoAppr |
|||
from module_admin.entity.do.data_ast_content_do import DataAssetInfo |
|||
from module_admin.entity.vo.dataast_vo import DataAstApprModel,DataAstInfoModel |
|||
from utils.page_util import PageUtil |
|||
from sqlalchemy.orm import aliased |
|||
from datetime import datetime |
|||
|
|||
|
|||
|
|||
class DataAstDao: |
|||
@classmethod |
|||
async def get_dataast_appr_list(cls, flowId:str,db: AsyncSession): |
|||
filters = [] |
|||
filters.append(DataAssetInfoAppr.flowId == flowId) |
|||
query = select(DataAssetInfoAppr).where(*filters).order_by(desc(DataAssetInfoAppr.create_time)) |
|||
return await PageUtil.paginate(db, query, 0, 0, False) |
|||
@classmethod |
|||
async def get_dataast_appr_list_Flow(cls, flowId:str,db: AsyncSession): |
|||
filters = [] |
|||
filters.append(DataAssetInfoAppr.flowId == flowId) |
|||
query = select(DataAssetInfoAppr).where(*filters).order_by(desc(DataAssetInfoAppr.create_time)) |
|||
return await PageUtil.paginate(db, query, 0, 0, False) |
|||
|
|||
|
|||
|
|||
@classmethod |
|||
async def add_dataast_appr(cls, db: AsyncSession, model: DataAstApprModel): |
|||
# 从model中获取ast_no |
|||
ast_no = model.ast_no |
|||
|
|||
# 查询当前ast_no对应的最大版本号 |
|||
from sqlalchemy import select, func |
|||
|
|||
query = select(func.max(DataAssetInfoAppr.version_no)).where( |
|||
DataAssetInfoAppr.ast_no == ast_no |
|||
) |
|||
|
|||
result = await db.execute(query) |
|||
max_version = result.scalar_one_or_none() |
|||
|
|||
# 计算新的版本号 |
|||
new_version = '1' if max_version is None else str(int(max_version) + 1) |
|||
|
|||
# 创建对象时设置版本号 |
|||
data = model.model_dump() |
|||
data['version_no'] = new_version |
|||
|
|||
col = DataAssetInfoAppr(**data) |
|||
db.add(col) |
|||
await db.flush() |
|||
|
|||
return col |
|||
|
|||
|
|||
|
|||
@classmethod |
|||
async def add_dataast_data(cls, db: AsyncSession, model: DataAstInfoModel): |
|||
# 子查询:获取每个 ast_no 对应的最新已成功审批版本 |
|||
subquery = ( |
|||
select( |
|||
DataAssetInfoAppr.ast_no, |
|||
func.max(DataAssetInfoAppr.version_no).label("max_version") |
|||
) |
|||
.filter( |
|||
DataAssetInfoAppr.approStatus == "succeed" |
|||
) |
|||
.group_by(DataAssetInfoAppr.ast_no) |
|||
.subquery() |
|||
) |
|||
|
|||
# 主查询:筛选符合条件的记录并映射到目标模型 |
|||
query = ( |
|||
select(DataAssetInfoAppr) |
|||
.join( |
|||
subquery, |
|||
and_( |
|||
DataAssetInfoAppr.ast_no == subquery.c.ast_no, |
|||
DataAssetInfoAppr.version_no == subquery.c.max_version |
|||
) |
|||
) |
|||
.filter( |
|||
DataAssetInfoAppr.data_ast_stat == "1", |
|||
DataAssetInfoAppr.changeType == "add", |
|||
DataAssetInfoAppr.approStatus == "succeed" |
|||
) |
|||
) |
|||
|
|||
# 执行查询获取符合条件的记录 |
|||
approved_records = await db.scalars(query) |
|||
|
|||
# 收集需要处理的记录数据 |
|||
records_to_upsert = [] |
|||
for record in approved_records: |
|||
# 从源记录中提取需要的字段 |
|||
asset_data = { |
|||
"data_ast_eng_name": record.data_ast_eng_name, |
|||
"data_ast_cn_name": record.data_ast_cn_name, |
|||
"data_ast_type": record.data_ast_type, |
|||
"data_ast_stat": record.data_ast_stat, |
|||
"data_ast_desc": record.data_ast_desc, |
|||
"data_ast_screen": record.data_ast_screen, |
|||
"data_ast_scren_clas": record.data_ast_scren_clas, |
|||
"data_ast_cont": record.data_ast_cont, |
|||
"data_ast_faq": record.data_ast_faq, |
|||
"data_ast_estb_time": record.data_ast_estb_time, |
|||
"data_ast_upd_time": datetime.now(), # 使用当前时间 |
|||
"data_ast_src": record.data_ast_src, |
|||
"ast_no": record.ast_no, |
|||
"data_ast_clas": record.data_ast_clas |
|||
} |
|||
records_to_upsert.append(asset_data) |
|||
|
|||
# 使用 bulk_insert_mappings 执行批量插入/更新 |
|||
if records_to_upsert: |
|||
from sqlalchemy.dialects.mysql import insert |
|||
|
|||
insert_stmt = insert(DataAssetInfo).values(records_to_upsert) |
|||
|
|||
# 定义 ON DUPLICATE KEY UPDATE 子句 |
|||
update_dict = { |
|||
c.name: insert_stmt.inserted[c.name] |
|||
for c in DataAssetInfo.__table__.columns |
|||
if c.name != 'data_ast_no' and c.name != 'ast_no' # 排除主键和业务主键 |
|||
} |
|||
# 手动设置 upd_time 为当前时间 |
|||
update_dict['data_ast_upd_time'] = datetime.now() |
|||
|
|||
upsert_stmt = insert_stmt.on_duplicate_key_update(update_dict) |
|||
|
|||
# 执行插入/更新语句 |
|||
await db.execute(upsert_stmt) |
|||
|
|||
# 提交会话使更改生效 |
|||
await db.commit() |
|||
|
|||
return "元数据成功发布到数据资产!" |
|||
|
|||
|
|||
|
|||
|
|||
@classmethod |
|||
async def update_data_ast_appr(cls, db: AsyncSession, update_data: DataAstInfoModel): |
|||
|
|||
await db.execute(update(DataAssetInfoAppr), [update_data]) |
|||
await db.flush() |
|||
|
|||
|
|||
@classmethod |
|||
async def get_ast_main_appr_list(cls, flowId:str,db: AsyncSession): |
|||
filters = [] |
|||
filters.append(DataAssetInfoAppr.flowId == flowId) |
|||
query = select(DataAssetInfoAppr).where(*filters).order_by(desc(DataAssetInfoAppr.create_time)) |
|||
return await PageUtil.paginate(db, query, 0, 0, False) |
@ -0,0 +1,67 @@ |
|||
|
|||
from sqlalchemy import Column, String, Integer, Text, DateTime |
|||
from sqlalchemy.sql import func |
|||
from config.database import Base |
|||
|
|||
class DataAssetInfoAppr(Base): |
|||
__tablename__ = 't_data_ast_info_appr' |
|||
|
|||
onum = Column(String(255), primary_key=True, comment='数据资产序号') |
|||
data_ast_eng_name = Column(String(255), nullable=True, comment='数据资产英文名称') |
|||
data_ast_cn_name = Column(String(255), nullable=True, comment='数据资产中文名称') |
|||
data_ast_type = Column(String(50), nullable=True, comment='数据资产类型') |
|||
data_ast_stat = Column(String(50), nullable=True, comment='数据资产状态(1:有效 0:无效)') |
|||
data_ast_desc = Column(Text, nullable=True, comment='数据资产描述/说明') |
|||
data_ast_screen = Column(String(255), nullable=True, comment='资产应用场景') |
|||
data_ast_scren_clas = Column(String(255), nullable=True, comment='应用场景分类') |
|||
data_ast_cont = Column(Text, nullable=True, comment='数据资产内容') |
|||
data_ast_faq = Column(Text, nullable=True, comment='数据资产常见问题') |
|||
data_ast_estb_time = Column(DateTime, nullable=True, server_default=func.now(), comment='数据资产建立时间') |
|||
data_ast_upd_time = Column(DateTime, nullable=True, server_default=func.now(), onupdate=func.now(), comment='数据资产更新时间') |
|||
data_ast_src = Column(String(255), nullable=True, comment='数据资产来源') |
|||
ast_no = Column(Integer, nullable=True, comment='数据资产编号') |
|||
data_ast_clas = Column(String(255), nullable=True, comment='数据资产标签') |
|||
version_no = Column(String(50), nullable=True, comment='版本号') |
|||
changeType = Column(String(10), default=None, comment='变更类型') |
|||
flowId = Column(String(50), default=None, comment='审批Id') |
|||
oldInstId = Column(String(50), default=None, comment='原始正式id') |
|||
compareId = Column(String(50), default=None, comment='原始数据id(用于对比差异)') |
|||
approStatus = Column(String(10), default=None, comment='审批状态(waiting,申请中,pending审核中,succeed,rejected已审核,canceled已取消)') |
|||
create_by = Column(String(255), nullable=True, comment='创建人') |
|||
create_time = Column(DateTime, nullable=True, server_default=func.now(), comment='创建时间') |
|||
upd_prsn = Column(String(255), nullable=True, comment='更新人') |
|||
upd_time = Column(DateTime, nullable=True, server_default=func.now(), onupdate=func.now(), comment='更新时间') |
|||
|
|||
|
|||
# class MetadataSuppInfo(Base): |
|||
# __tablename__ = 't_metadata_supp_info' |
|||
|
|||
# onum = Column(String(36), primary_key=True, comment='唯一编号') |
|||
# crrct_ver_num = Column(String(50), comment='补录版本号') |
|||
# ssys_cd = Column(String(50), comment='系统代码') |
|||
# mdl_name = Column(String(50), comment='模型名称') |
|||
# tab_eng_name = Column(String(250), comment='表英文名称') |
|||
# tab_crrct_name = Column(String(250), comment='表补录名称') |
|||
# tab_desc = Column(String(500), comment='表描述') |
|||
# pic = Column(String(64), comment='表图片') |
|||
# gov_flag = Column(String(1), comment='治理标志(0否 1是)') |
|||
# rec_stat = Column(String(1), comment='记录状态(0有效 1无效)') |
|||
# tab_clas = Column(Text, comment='表分类') |
|||
# rec_subm_prsn = Column(String(255), comment='记录提交人') |
|||
# upd_time = Column(DateTime, default=func.now(), onupdate=func.now(), comment='更新时间') |
|||
|
|||
# class MetadataExtractInfo(Base): |
|||
# __tablename__ = 't_metadata_extract_info' |
|||
|
|||
# onum = Column(Integer, primary_key=True, comment='唯一编号') |
|||
# extract_ver_num = Column(String(50), comment='采集版本号') |
|||
# ver_desc = Column(String(250), comment='版本描述') |
|||
# ssys_cd = Column(String(50), comment='系统代码') |
|||
# data_whs_name = Column(String(50), comment='数据仓库名称') |
|||
# mdl_name = Column(String(50), comment='模式名称') |
|||
# tab_no = Column(Integer, comment='表编号') |
|||
# tab_type = Column(String(50), comment='表类型') |
|||
# tab_eng_name = Column(String(250), comment='表英文名称') |
|||
# tab_cn_name = Column(String(250), comment='表中文名称') |
|||
# tab_rec_num = Column(Integer, comment='记录数') |
|||
# upd_time = Column(DateTime, default=func.now(), onupdate=func.now(), comment='更新时间') |
@ -0,0 +1,69 @@ |
|||
from pydantic import BaseModel |
|||
from typing import Optional,List |
|||
from datetime import datetime |
|||
from pydantic.alias_generators import to_camel |
|||
from module_admin.annotation.pydantic_annotation import as_query |
|||
from pydantic import ConfigDict, Field |
|||
|
|||
|
|||
class DataAstInfoModel(BaseModel): |
|||
""" |
|||
数据资产信息模型 (对应 t_data_ast_info_appr 表) |
|||
""" |
|||
model_config = ConfigDict(alias_generator=to_camel, from_attributes=True) |
|||
|
|||
class DataAstInfoModel(BaseModel): |
|||
""" |
|||
数据资产信息模型 (对应 t_data_ast_info_appr 表) |
|||
""" |
|||
model_config = ConfigDict(alias_generator=to_camel, from_attributes=True) |
|||
|
|||
onum: Optional[str] = Field(default=None, description='序号', alias='onum') |
|||
data_ast_eng_name: Optional[str] = Field(default=None, description='数据资产英文名称', alias='dataAstEngName') |
|||
data_ast_cn_name: Optional[str] = Field(default=None, description='数据资产中文名称', alias='dataAstCnName') |
|||
data_ast_type: Optional[str] = Field(default=None, description='数据资产类型', alias='dataAstType') |
|||
data_ast_stat: Optional[str] = Field(default=None, description='数据资产状态', alias='dataAstStat') |
|||
data_ast_desc: Optional[str] = Field(default=None, description='数据资产描述/说明', alias='dataAstDesc') |
|||
data_ast_screen: Optional[str] = Field(default=None, description='资产应用场景', alias='dataAstScreen') |
|||
data_ast_scren_clas: Optional[str] = Field(default=None, description='应用场景分类', alias='dataAstScrenClas') |
|||
data_ast_cont: Optional[str] = Field(default=None, description='数据资产内容', alias='dataAstCont') |
|||
data_ast_faq: Optional[str] = Field(default=None, description='数据资产常见问题', alias='dataAstFaq') |
|||
data_ast_estb_time: Optional[datetime] = Field(default=None, description='数据资产建立时间', alias='dataAstEstbTime') |
|||
data_ast_upd_time: Optional[datetime] = Field(default=None, description='数据资产更新时间', alias='dataAstUpdTime') |
|||
data_ast_src: Optional[str] = Field(default=None, description='数据资产来源', alias='dataAstSrc') |
|||
ast_no: Optional[int] = Field(default=None, description='数据资产编号', alias='astNo') |
|||
data_ast_clas: Optional[str] = Field(default=None, description='数据资产标签', alias='dataAstClas') |
|||
version_no: Optional[str] = Field(default=None, description='版本号', alias='versionNo') |
|||
create_by: Optional[str] = Field(default=None, description='创建人', alias='createBy') |
|||
create_time: Optional[datetime] = Field(default=None, description='创建时间', alias='createTime') |
|||
upd_prsn: Optional[str] = Field(default=None, description='更新者', alias='updPrsn') |
|||
upd_time: Optional[datetime] = Field(default=None, description='更新时间', alias='updTime') |
|||
|
|||
|
|||
|
|||
@as_query |
|||
class DataAstInfoPageQueryModel(DataAstInfoModel): |
|||
""" |
|||
数据资产分页查询模型 |
|||
""" |
|||
page_num: int = Field(default=1, description='当前页码') |
|||
page_size: int = Field(default=10, description='每页记录数') |
|||
|
|||
|
|||
class DataAstApprModel(DataAstInfoModel): |
|||
""" |
|||
数据资产审批申请模型 |
|||
""" |
|||
changeType: Optional[str] = Field(default=None, description='变更类型') |
|||
flowId: Optional[str] = Field(default=None, description='审批Id') |
|||
oldInstId: Optional[str] = Field(default=None, description='原始正式id') |
|||
compareId: Optional[str] = Field(default=None, description='原始数据id(用于对比差异)') |
|||
approStatus: Optional[str] = Field(default=None, description='审批状态(waiting,申请中,pending审核中,succeed,rejected已审核,canceled已取消)') |
|||
|
|||
|
|||
|
|||
class DataAstApprBatchModel(BaseModel): |
|||
""" |
|||
数据资产审批申请批量提交模型 |
|||
""" |
|||
assetItems: List[DataAstApprModel] |
@ -0,0 +1,70 @@ |
|||
from fastapi import Request |
|||
from sqlalchemy.ext.asyncio import AsyncSession |
|||
from exceptions.exception import ServiceException |
|||
from module_admin.dao.dataast_dao import DataAstDao |
|||
from module_admin.entity.vo.dataast_vo import DataAstApprModel,DataAstInfoModel,DataAstApprBatchModel |
|||
from utils.common_util import CamelCaseUtil |
|||
import uuid |
|||
from module_admin.entity.vo.approval_vo import ApplyModel |
|||
from module_admin.service.approval_service import ApprovalService |
|||
from module_admin.entity.vo.common_vo import CrudResponseModel |
|||
|
|||
|
|||
|
|||
class DataAstService: |
|||
""" |
|||
数据源标准服务层 |
|||
""" |
|||
@classmethod |
|||
async def get_dataast_appr_list(cls, query_db: AsyncSession, query_object: DataAstApprModel, is_page: bool = False): |
|||
return await DataAstDao.get_dataast_appr_list(query_object.flowId, query_db) |
|||
|
|||
|
|||
|
|||
# @classmethod |
|||
# async def add_dataasts_appr(cls, query_db: AsyncSession, model: DataAstInfoModel): |
|||
# # if not await cls.check_dict_unique_services(query_db, model): |
|||
# # raise ServiceException(message=f"数据资产编号 {model.ast_no} 已存在") |
|||
# model.onum=str(uuid.uuid4()) |
|||
# model.data_ast_stat="1" |
|||
# apprModel = DataAstApprModel(**model.model_dump(exclude_unset=True, by_alias=True)) |
|||
# apprModel.changeType="add" |
|||
# apprModel.compareId=model.onum |
|||
# apprModel.oldInstId=model.onum |
|||
# apprModel.approStatus="waiting" |
|||
# apprModel.flowId=str(uuid.uuid4()) |
|||
# await DataAstDao.add_dataast_appr(query_db, apprModel) |
|||
# applyModel = ApplyModel() |
|||
# applyModel.businessType = "dataAssetMain" |
|||
# applyModel.businessId = apprModel.flowId |
|||
# applyModel.applicant = apprModel.create_by |
|||
# await ApprovalService.apply_services(query_db, applyModel, 'dataAssetMain') |
|||
# return CrudResponseModel(is_success=True, message='新增数据资产成功') |
|||
|
|||
@classmethod |
|||
async def add_dataasts_appr(cls, query_db: AsyncSession, models: DataAstApprBatchModel,username: str): |
|||
# if not await cls.check_dict_unique_services(query_db, model): |
|||
# raise ServiceException(message=f"数据资产编号 {model.ast_no} 已存在") |
|||
flowId = str(uuid.uuid4()) #流程编号 |
|||
for model in models.assetItems: |
|||
model.onum=str(uuid.uuid4()) |
|||
model.data_ast_stat="1" |
|||
apprModel = DataAstApprModel(**model.model_dump(exclude_unset=True, by_alias=True)) |
|||
apprModel.changeType="add" |
|||
apprModel.compareId=model.onum |
|||
apprModel.oldInstId=model.onum |
|||
apprModel.approStatus="waiting" |
|||
apprModel.flowId=flowId |
|||
|
|||
await DataAstDao.add_dataast_appr(query_db, apprModel) |
|||
applyModel = ApplyModel() |
|||
applyModel.businessType = "dataAssetMain" |
|||
applyModel.businessId = flowId |
|||
applyModel.applicant = username |
|||
await ApprovalService.apply_services(query_db, applyModel, 'dataAssetMain') |
|||
return CrudResponseModel(is_success=True, message='新增数据资产成功') |
|||
|
|||
|
|||
@classmethod |
|||
async def get_ast_main_appr_list(cls, query_db: AsyncSession, query_object: DataAstInfoModel, is_page: bool = False): |
|||
return await DataAstDao.get_ast_main_appr_list(query_object.flowId, query_db) |
Loading…
Reference in new issue