You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
737 lines
30 KiB
737 lines
30 KiB
from sqlalchemy import delete, select, update, desc,or_,not_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from module_admin.entity.do.datastd_do import DataStdCode,DataStdDict,DataAstContent,DataAstContentRela,DataStdMain
|
|
from module_admin.entity.vo.datastd_vo import DataStdCodeModel,DataStdDictModel,DataStdMainModel
|
|
from module_admin.entity.do.dept_do import SysDept
|
|
from utils.page_util import PageUtil
|
|
from sqlalchemy.orm import aliased
|
|
from module_admin.entity.vo.data_ast_content_vo import DataCatalogPageQueryModel, DeleteDataCatalogModel,DataCatalogChild
|
|
from datetime import datetime
|
|
|
|
|
|
class DataStdDao:
|
|
|
|
# ----------------------------------------------------------------数据标准模块----------------------------------------------------------------------------------------------------
|
|
|
|
@classmethod
|
|
async def get_std_code_list(cls, db: AsyncSession, query_object: DataStdCodeModel, is_page: bool = False):
|
|
"""
|
|
获取 DataStdCode 的列表信息，支持模糊查询和分页
|
|
:param db: ORM对象
|
|
:param query_object: 查询参数对象
|
|
:param is_page: 是否开启分页
|
|
:return: 列表信息
|
|
"""
|
|
# 构建查询条件
|
|
filters = []
|
|
if query_object.code_name:
|
|
filters.append(DataStdCode.code_name.like(f"%{query_object.code_name}%"))
|
|
if query_object.code_num:
|
|
filters.append( or_(
|
|
DataStdCode.code_name.like(f"%{query_object.code_num}%"),
|
|
DataStdCode.code_num.like(f"%{query_object.code_num}%")
|
|
))
|
|
if query_object.code_status:
|
|
filters.append(DataStdCode.code_status==query_object.code_status)
|
|
if query_object.sys_id:
|
|
filters.append(DataStdCode.sys_id==query_object.sys_id)
|
|
if query_object.code_type:
|
|
filters.append(DataStdCode.code_type==query_object.code_type)
|
|
|
|
if query_object.class_id:
|
|
filters.append(DataStdCode.class_id==query_object.class_id)
|
|
if query_object.class_id=='codeItem' and not query_object.parent_id:
|
|
filters.append(1==2)
|
|
if query_object.parent_id:
|
|
filters.append(DataStdCode.parent_id==query_object.parent_id)
|
|
# 构建查询语句
|
|
query = (
|
|
select(DataStdCode)
|
|
.where(*filters)
|
|
.order_by(desc(DataStdCode.create_time)) # 按创建时间降序排序
|
|
)
|
|
|
|
# 分页处理
|
|
col_list = await PageUtil.paginate(
|
|
db, query, query_object.page_num, query_object.page_size, is_page
|
|
)
|
|
return col_list
|
|
@classmethod
|
|
async def get_std_code_map_list(cls, db: AsyncSession, query_object: DataStdCodeModel, is_page: bool = False):
|
|
# 构建查询条件
|
|
filters = []
|
|
# 系统级代码
|
|
c1 = aliased(DataStdCode) # 当前 codeItem 表（c1）
|
|
c2 = aliased(DataStdCode) # 别名 c2 表示父级（parent）
|
|
c3 = aliased(DataStdCode) # 别名 c3 表示映射的代码
|
|
c4 = aliased(DataStdCode) # 别名 c4 表示映射父级代码
|
|
if query_object.parent_id:
|
|
if query_object.code_type =='sys':
|
|
filters.append(c1.parent_id == query_object.parent_id)
|
|
else:
|
|
filters.append(c3.parent_id == query_object.parent_id)
|
|
# 公司级代码
|
|
# 2. 创建别名并进行连接
|
|
filters.append(c1.class_id == 'codeItem') # class_id 为 'codeItem'
|
|
# 3. 构建基础查询，使用连接和别名
|
|
query = (
|
|
select(
|
|
c2.sys_id.label("target_sys_id"), # 父级系统编号
|
|
c2.sys_name.label("target_sys_name"), # 父级系统名称
|
|
c2.code_num.label("target_code_num"), # 父级代码编号
|
|
c2.code_name.label("target_code_name"), # 父级代码名称
|
|
c1.code_num.label("target_code_num_item"), # 当前 codeItem 的编号
|
|
c1.code_name.label("target_code_name_item"), # 当前 codeItem 的名称
|
|
c4.code_num.label("resource_code_num"), # 映射的父级代码编号
|
|
c4.code_name.label("resource_code_name"), # 映射的父级代码名称
|
|
c3.code_num.label("resource_code_num_item"), # 映射的父级代码编号
|
|
c3.code_name.label("resource_code_name_item") # 映射的父级代码名称
|
|
)
|
|
.select_from(c1) # 从 c1 开始查询
|
|
.join(c2, c1.parent_id == c2.id, isouter=True) # 连接 c2
|
|
.join(c3, c1.code_map_id == c3.id, isouter=True) # 连接 c3
|
|
.join(c4, c3.parent_id == c4.id, isouter=True) # 连接 c4
|
|
.where(*filters) # 使用所有过滤条件
|
|
.order_by(desc(c1.create_time)) # 按照创建时间降序排列
|
|
)
|
|
# 4. 分页逻辑
|
|
col_list = await PageUtil.paginate(
|
|
db, query, query_object.page_num, query_object.page_size, is_page
|
|
)
|
|
return col_list
|
|
@classmethod
|
|
async def get_data_code_list_by_info(cls, db: AsyncSession, query_object: DataStdCodeModel):
|
|
List = (
|
|
await db.execute(
|
|
select(DataStdCode)
|
|
.where(
|
|
DataStdCode.code_name == query_object.code_name if query_object.code_name else True,
|
|
DataStdCode.code_status == query_object.code_status if query_object.code_status else True,
|
|
DataStdCode.sys_id == query_object.sys_id if query_object.sys_id else True,
|
|
DataStdCode.code_type == query_object.code_type if query_object.code_type else True,
|
|
DataStdCode.parent_id == query_object.parent_id if query_object.parent_id else True,
|
|
DataStdCode.class_id == query_object.class_id if query_object.class_id else True
|
|
)
|
|
)
|
|
).scalars().all()
|
|
return List
|
|
@classmethod
|
|
async def get_data_code_by_info(cls, db: AsyncSession, query_object: DataStdCodeModel):
|
|
List = (
|
|
await db.execute(
|
|
select(DataStdCode)
|
|
.where(
|
|
DataStdCode.code_name == query_object.code_name if query_object.code_name else True,
|
|
DataStdCode.code_num == query_object.code_num if query_object.code_num else True,
|
|
DataStdCode.code_status == query_object.code_status if query_object.code_status else True,
|
|
DataStdCode.sys_id == query_object.sys_id if query_object.sys_id else True,
|
|
DataStdCode.code_type == query_object.code_type if query_object.code_type else True,
|
|
DataStdCode.parent_id == query_object.parent_id if query_object.parent_id else True,
|
|
DataStdCode.class_id == query_object.class_id if query_object.class_id else True
|
|
)
|
|
)
|
|
).scalars().first()
|
|
return List
|
|
@classmethod
|
|
async def get_std_code_by_id(cls, db: AsyncSession, Id: str):
|
|
col = (
|
|
await db.execute(
|
|
select(DataStdCode)
|
|
.where(DataStdCode.id == Id)
|
|
)
|
|
).scalars().first()
|
|
return col
|
|
|
|
|
|
@classmethod
|
|
async def add_std_code(cls, db: AsyncSession,model:DataStdCodeModel):
|
|
col = DataStdCode(
|
|
**model.model_dump()
|
|
)
|
|
db.add(col)
|
|
await db.flush()
|
|
return col
|
|
|
|
@classmethod
|
|
async def delete_std_code(cls, db: AsyncSession, Id: str):
|
|
await db.execute(delete(DataStdCode).where(DataStdCode.id == Id))
|
|
|
|
|
|
@classmethod
|
|
async def update_std_code(cls, db: AsyncSession, update_data: DataStdCodeModel):
|
|
|
|
await db.execute(update(DataStdCode), [update_data])
|
|
await db.flush()
|
|
|
|
@classmethod
|
|
async def check_code_num_exists(cls, query_db: AsyncSession, code_num: str) -> bool:
|
|
"""
|
|
检查标准代码或代码项的code_num是否已经存在
|
|
"""
|
|
# 查询标准代码表中是否存在相同的code_num
|
|
result = await query_db.execute(select(DataStdCodeModel).filter(DataStdCodeModel.code_num == code_num))
|
|
existing_code = result.scalar_one_or_none()
|
|
|
|
# 如果不存在，则检查代码项表中是否存在相同的code_num
|
|
if not existing_code:
|
|
result = await query_db.execute(select(DataStdCode).filter(DataStdCodeModel.code_num == code_num))
|
|
existing_code_item = result.scalar_one_or_none()
|
|
return existing_code_item is not None
|
|
|
|
return True
|
|
# ----------------------------------------------------------------数据字典----------------------------------------------------------------------------------------------------
|
|
@classmethod
|
|
async def get_std_dict_list(cls, db: AsyncSession, query_object: DataStdDictModel, is_page: bool = False):
|
|
"""
|
|
获取 DataStdDict 的列表信息，支持模糊查询和分页
|
|
:param db: ORM对象
|
|
:param query_object: 查询参数对象
|
|
:param is_page: 是否开启分页
|
|
:return: 列表信息
|
|
"""
|
|
# 构建查询条件
|
|
filters = []
|
|
if query_object.dict_name:
|
|
filters.append(DataStdDict.dict_name.like(f"%{query_object.dict_name}%"))
|
|
if query_object.dict_code:
|
|
filters.append(DataStdDict.dict_code.like(f"%{query_object.dict_code}%"))
|
|
if query_object.dict_level:
|
|
filters.append(DataStdDict.dict_level == query_object.dict_level)
|
|
if query_object.sys_id:
|
|
filters.append(DataStdDict.sys_id == query_object.sys_id)
|
|
if query_object.dict_type:
|
|
filters.append(DataStdDict.dict_type == query_object.dict_type)
|
|
if query_object.dict_status:
|
|
filters.append(DataStdDict.dict_status == query_object.dict_status)
|
|
c1 = aliased(SysDept) # 技术部门
|
|
c2 = aliased(SysDept) # 业务部门
|
|
# 构建查询语句
|
|
query = (
|
|
select(DataStdDict.id,
|
|
DataStdDict.create_by,
|
|
DataStdDict.create_time,
|
|
DataStdDict.update_by,
|
|
DataStdDict.update_time,
|
|
DataStdDict.dict_num,
|
|
DataStdDict.dict_code,
|
|
DataStdDict.dict_name,
|
|
DataStdDict.dict_level,
|
|
DataStdDict.dict_type,
|
|
DataStdDict.sys_name,
|
|
DataStdDict.sys_id,
|
|
DataStdDict.dict_menu,
|
|
DataStdDict.data_type,
|
|
DataStdDict.std_code,
|
|
DataStdDict.std_name,
|
|
DataStdDict.dict_status,
|
|
DataStdDict.buss_dept_id,
|
|
DataStdDict.tech_dept_id,
|
|
DataStdDict.buss_user,
|
|
DataStdDict.tech_user,
|
|
c1.dept_name.label("tech_dept_name"),
|
|
c2.dept_name.label("buss_dept_name")).select_from(DataStdDict)
|
|
.join(c1,c1.dept_id ==DataStdDict.tech_dept_id, isouter=True)
|
|
.join(c2,c2.dept_id ==DataStdDict.buss_dept_id, isouter=True)
|
|
.where(*filters)
|
|
.order_by(desc(DataStdDict.create_time)) # 按创建时间降序排序
|
|
)
|
|
|
|
# 分页处理
|
|
col_list = await PageUtil.paginate(
|
|
db, query, query_object.page_num, query_object.page_size, is_page
|
|
)
|
|
|
|
return col_list
|
|
@classmethod
|
|
async def add_std_dict(cls, db: AsyncSession,model:DataStdDictModel):
|
|
col = DataStdDict(
|
|
**model.model_dump()
|
|
)
|
|
db.add(col)
|
|
await db.flush()
|
|
return col
|
|
|
|
@classmethod
|
|
async def delete_std_dict(cls, db: AsyncSession, Id: str):
|
|
await db.execute(delete(DataStdDict).where(DataStdDict.id == Id))
|
|
@classmethod
|
|
async def get_std_dict_list_all(cls, db: AsyncSession, query_object: DataStdDict):
|
|
filters = []
|
|
if query_object.std_code:
|
|
filters.append(DataStdDict.std_code.like(f"%{query_object.std_code}%"))
|
|
query = select(DataStdDict).where(*filters).order_by(desc(DataStdDict.create_time))
|
|
return await PageUtil.paginate(db, query, 0, 0, False)
|
|
@classmethod
|
|
async def update_std_dict(cls, db: AsyncSession, update_data: DataStdDictModel):
|
|
|
|
await db.execute(update(DataStdDict), [update_data])
|
|
await db.flush()
|
|
|
|
@classmethod
|
|
async def get_std_dict_by_id(cls, db: AsyncSession, Id: str):
|
|
col = (
|
|
await db.execute(
|
|
select(DataStdDict)
|
|
.where(DataStdDict.id == Id)
|
|
)
|
|
).scalars().first()
|
|
return col
|
|
@classmethod
|
|
async def get_data_dict_by_info(cls, db: AsyncSession, query_object: DataStdDictModel):
|
|
List = (
|
|
await db.execute(
|
|
select(DataStdDict)
|
|
.where(
|
|
DataStdDict.dict_name == query_object.dict_name if query_object.dict_name else True,
|
|
DataStdDict.dict_num == query_object.dict_num if query_object.dict_num else True,
|
|
DataStdDict.dict_status == query_object.dict_status if query_object.dict_status else True,
|
|
DataStdDict.std_code == query_object.std_code if query_object.std_code else True,
|
|
DataStdDict.sys_id == query_object.sys_id if query_object.sys_id else True,
|
|
DataStdDict.dict_type == query_object.dict_type if query_object.dict_type else True,
|
|
)
|
|
)
|
|
).scalars().first()
|
|
return List
|
|
# 数据标准目录
|
|
|
|
@classmethod
|
|
async def get_catalog_by_id(cls, db: AsyncSession, content_onum: int):
|
|
"""
|
|
根据目录ID获取目录详细信息
|
|
|
|
:param db: orm对象
|
|
:param content_onum: 目录ID
|
|
:return: 目录信息对象
|
|
"""
|
|
catalog_info = (
|
|
(await db.execute(select(DataAstContent).where(DataAstContent.content_onum == content_onum , DataAstContent.content_stat == 1)))
|
|
.scalars()
|
|
.first()
|
|
)
|
|
|
|
return catalog_info
|
|
|
|
@classmethod
|
|
async def get_catalog_detail_by_info(cls, db: AsyncSession, catalog: DataCatalogPageQueryModel):
|
|
"""
|
|
根据目录参数获取目录信息
|
|
|
|
:param db: orm对象
|
|
:param catalog: 目录参数对象
|
|
:return: 目录信息对象
|
|
"""
|
|
catalog_info = (
|
|
(
|
|
await db.execute(
|
|
select(DataAstContent).where(
|
|
DataAstContent.content_name == catalog.content_name if catalog.content_name else True,
|
|
DataAstContent.content_stat == catalog.content_stat if catalog.content_stat else True,
|
|
DataAstContent.content_pic == catalog.content_pic if catalog.content_pic else True,
|
|
DataAstContent.content_stat == 1,
|
|
)
|
|
)
|
|
)
|
|
.scalars()
|
|
.first()
|
|
)
|
|
|
|
return catalog_info
|
|
|
|
@classmethod
|
|
async def update_leaf_node_flag(cls, db: AsyncSession):
|
|
"""
|
|
更新leaf_node_flag字段
|
|
"""
|
|
# 创建别名对象
|
|
t2 = aliased(DataAstContent, name='t2') # 正确使用aliased创建别名
|
|
subquery = (
|
|
select(DataAstContent.content_onum)
|
|
.where(
|
|
DataAstContent.content_stat == '1',
|
|
DataAstContent.leaf_node_flag == 0,
|
|
not_(
|
|
select(1)
|
|
.select_from(t2) # 使用别名后的表
|
|
.where(
|
|
t2.supr_content_onum == DataAstContent.content_onum,
|
|
t2.content_stat == '1'
|
|
)
|
|
.exists() # 添加exists()方法
|
|
)
|
|
)
|
|
).alias('temp')
|
|
|
|
stmt = (
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum.in_(subquery))
|
|
.values(leaf_node_flag=1, upd_time=datetime.now())
|
|
)
|
|
await db.execute(stmt)
|
|
|
|
@classmethod
|
|
async def add_catalog_dao(cls, db: AsyncSession, catalog1: dict, catalog2: dict):
|
|
"""
|
|
新增目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 目录对象
|
|
:return:
|
|
"""
|
|
db_catalog = DataAstContent(**catalog1)
|
|
db.add(db_catalog)
|
|
await db.flush()
|
|
|
|
|
|
# 处理子关系（统一转换为 ORM 模型）
|
|
for child in catalog2.get('children', []):
|
|
# 如果是 Pydantic 模型实例，先转换为字典
|
|
if isinstance(child, DataCatalogChild):
|
|
child_dict = child.model_dump()
|
|
elif isinstance(child, dict):
|
|
child_dict = child
|
|
else:
|
|
raise TypeError("不支持的子关系数据类型")
|
|
|
|
# 创建 ORM 模型实例
|
|
processed_child = dict(child_dict)
|
|
processed_child['content_onum'] = db_catalog.content_onum
|
|
db_child = DataAstContentRela(**processed_child)
|
|
|
|
db.add(db_child)
|
|
await db.flush()
|
|
|
|
return db_catalog
|
|
@classmethod
|
|
async def edit_catalog_leaf_dao(cls, db: AsyncSession, catalog: dict):
|
|
"""
|
|
编辑叶子节点目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
content_onum = catalog['content_onum']
|
|
stmt = (
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum == content_onum)
|
|
.values(
|
|
leaf_node_flag=catalog['leaf_node_flag']
|
|
)
|
|
)
|
|
|
|
await db.execute(stmt)
|
|
@classmethod
|
|
async def edit_catalog_child_dao(cls, db: AsyncSession, catalog: dict):
|
|
"""
|
|
编辑目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
content_onum = catalog['content_onum']
|
|
|
|
stmt = (
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum == content_onum)
|
|
.values(
|
|
content_name=catalog['content_name'],
|
|
content_stat=catalog['content_stat'],
|
|
content_intr=catalog['content_intr'],
|
|
content_pic=catalog['content_pic'],
|
|
supr_content_onum=catalog['supr_content_onum'],
|
|
leaf_node_flag=catalog['leaf_node_flag'],
|
|
upd_prsn=catalog['upd_prsn'],
|
|
upd_time=datetime.now()
|
|
) )
|
|
|
|
await db.execute(stmt)
|
|
|
|
# 处理子关系
|
|
for child in catalog.get('children', []):
|
|
rela_onum = child.get('rela_onum')
|
|
if rela_onum:
|
|
st = (
|
|
update(DataAstContentRela)
|
|
.where(DataAstContentRela.rela_onum == rela_onum)
|
|
.values(
|
|
content_onum=child.get('content_onum'),
|
|
ast_onum=child.get('ast_onum'),
|
|
rela_type=child.get('rela_type'),
|
|
rela_eff_begn_date=child.get('rela_eff_begn_date'),
|
|
rela_eff_end_date=child.get('rela_eff_end_date'),
|
|
upd_prsn=child.get('upd_prsn'))
|
|
)
|
|
|
|
await db.execute(st)
|
|
await cls.update_leaf_node_flag(db)
|
|
else:
|
|
child['content_onum'] = content_onum
|
|
db_child = DataAstContentRela(**child)
|
|
db.add(db_child)
|
|
await db.flush()
|
|
await cls.update_leaf_node_flag(db)
|
|
|
|
@classmethod
|
|
async def delete_catalog_dao(cls, db: AsyncSession, catalog: DeleteDataCatalogModel):
|
|
"""
|
|
删除目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 目录对象
|
|
:content_stat=0 作废
|
|
:return:
|
|
"""
|
|
content_onums = catalog.content_onums.split(',')
|
|
await db.execute(
|
|
update(DataAstContentRela)
|
|
.where(DataAstContentRela.content_onum.in_(content_onums))
|
|
.values(
|
|
rela_status=0,
|
|
rela_eff_end_date=datetime.now()
|
|
)
|
|
)
|
|
await db.execute(
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum.in_(content_onums))
|
|
.values(
|
|
content_stat=0,
|
|
upd_time=datetime.now()
|
|
)
|
|
)
|
|
|
|
|
|
await cls.update_leaf_node_flag(db)
|
|
|
|
@classmethod
|
|
async def moved_catalog_instr_dao(cls, db: AsyncSession, moved_catalog_data: dict):
|
|
"""
|
|
编辑目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
# content_onum = moved_catalog_data['content_onum']
|
|
|
|
stmt = (
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum == moved_catalog_data['content_onum'] , DataAstContent.supr_content_onum == moved_catalog_data['supr_content_onum'])
|
|
.values(
|
|
supr_content_onum=moved_catalog_data['supr_content_onum_after'],
|
|
upd_time=datetime.now()
|
|
) )
|
|
|
|
await db.execute(stmt)
|
|
await cls.update_leaf_node_flag(db)
|
|
|
|
@classmethod
|
|
async def merge_catalog_instr_dao(cls, db: AsyncSession, merge_catalog_data: dict):
|
|
"""
|
|
编辑目录数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
|
|
# stmt = (
|
|
# update(DataAstContent)
|
|
# .where(DataAstContent.content_onum == merge_catalog_data['content_onum'] , DataAstContent.supr_content_onum == merge_catalog_data['supr_content_onum'])
|
|
# .values(
|
|
# content_onum=merge_catalog_data['content_onum_after'],
|
|
# supr_content_onum=merge_catalog_data['supr_content_onum_after'],
|
|
# upd_time=datetime.now()
|
|
# ) )
|
|
|
|
# await db.execute(stmt)
|
|
stmt1 = (
|
|
update(DataAstContentRela)
|
|
.where( DataAstContentRela.content_onum == merge_catalog_data['content_onum'] and DataAstContentRela.rela_status == 1 )
|
|
.values(
|
|
content_onum=merge_catalog_data['content_onum_after'],
|
|
rela_eff_begn_date=datetime.now()
|
|
)
|
|
)
|
|
await db.execute(stmt1)
|
|
|
|
stmt2 = (
|
|
update(DataAstContent)
|
|
.where(DataAstContent.content_onum == merge_catalog_data['content_onum'] , DataAstContent.supr_content_onum == merge_catalog_data['supr_content_onum'])
|
|
.values( content_stat = '0' )
|
|
)
|
|
await db.execute(stmt2)
|
|
|
|
await cls.update_leaf_node_flag(db)
|
|
@classmethod
|
|
async def removerel_data_ast_catalog_dao(cls, db: AsyncSession, removerel_catalog_data: dict):
|
|
"""
|
|
编辑资产关系数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
|
|
stmt = (
|
|
update(DataAstContentRela)
|
|
.where(DataAstContentRela.rela_onum == removerel_catalog_data['rela_onum'] , DataAstContentRela.content_onum == removerel_catalog_data['content_onum'])
|
|
.values(
|
|
rela_status=removerel_catalog_data['rela_status']
|
|
) )
|
|
|
|
await db.execute(stmt)
|
|
await cls.update_leaf_node_flag(db)
|
|
|
|
|
|
@classmethod
|
|
async def moverel_data_ast_catalog_dao(cls, db: AsyncSession, moverel_catalog_data: dict):
|
|
"""
|
|
编辑资产关系数据库操作
|
|
|
|
:param db: orm对象
|
|
:param catalog: 需要更新的目录字典
|
|
:return:
|
|
"""
|
|
|
|
stmt = (
|
|
update(DataAstContentRela)
|
|
.where(DataAstContentRela.rela_onum == moverel_catalog_data['rela_onum'] , DataAstContentRela.content_onum == moverel_catalog_data['content_onum'])
|
|
.values(
|
|
content_onum=moverel_catalog_data['content_onum_after'],
|
|
rela_eff_end_date=datetime.now()
|
|
) )
|
|
|
|
await db.execute(stmt)
|
|
await cls.update_leaf_node_flag(db)
|
|
@classmethod
|
|
async def get_catalog_list(cls, db: AsyncSession, query_object: DataCatalogPageQueryModel, user_id: int, is_page: bool = False):
|
|
"""
|
|
根据查询参数获取数据资产目录列表
|
|
|
|
:param db: 异步会话对象
|
|
:param query_object: 分页查询参数对象
|
|
:param is_page: 是否分页
|
|
:return: 数据资产目录分页列表
|
|
"""
|
|
|
|
|
|
# 修改子查询部分
|
|
subquery_t1 = (
|
|
select(DataAstContentRela)
|
|
.where(DataAstContentRela.upd_prsn == query_object.upd_prsn, DataAstContentRela.content_onum == '2' and DataAstContentRela.rela_status == '1')
|
|
.union_all(
|
|
select(DataAstContentRela)
|
|
.where(DataAstContentRela.content_onum != '2' and DataAstContentRela.rela_status == '1')
|
|
)
|
|
).alias('subquery_t1') # 为子查询分配唯一别名
|
|
|
|
query = (
|
|
select(
|
|
DataAstContent.content_onum,
|
|
DataAstContent.content_name,
|
|
DataAstContent.content_stat,
|
|
DataAstContent.content_intr,
|
|
DataAstContent.content_pic,
|
|
DataAstContent.supr_content_onum,
|
|
DataAstContent.leaf_node_flag,
|
|
DataAstContent.upd_prsn,
|
|
DataAstContent.upd_time,
|
|
|
|
subquery_t1.c.rela_onum, # 明确指定子查询的字段
|
|
subquery_t1.c.ast_onum,
|
|
subquery_t1.c.rela_type,
|
|
subquery_t1.c.rela_eff_begn_date,
|
|
subquery_t1.c.rela_eff_end_date,
|
|
subquery_t1.c.upd_prsn,
|
|
)
|
|
.distinct()
|
|
.select_from(DataAstContent)
|
|
.outerjoin(subquery_t1, DataAstContent.content_onum == subquery_t1.c.content_onum) # 明确使用子查询别名
|
|
.where(DataAstContent.content_stat == 1)
|
|
.order_by(DataAstContent.content_onum)
|
|
)
|
|
|
|
# 使用分页工具进行查询
|
|
data_ast_list = await PageUtil.paginate(
|
|
db,
|
|
query,
|
|
page_num=query_object.page_num,
|
|
page_size=query_object.page_size,
|
|
is_page=is_page
|
|
)
|
|
|
|
return data_ast_list
|
|
#------------------------------------------------------------数据标准(658行)-------------------------------------------------------------------------------------------------
|
|
@classmethod
|
|
async def get_std_main_list(cls, db: AsyncSession, query_object: DataStdMainModel, is_page: bool = False):
|
|
filters = []
|
|
if query_object.std_name:
|
|
filters.append(DataStdMain.std_name.like(f"%{query_object.std_name}%"))
|
|
if query_object.std_code:
|
|
filters.append(DataStdMain.std_code.like(f"%{query_object.std_code}%"))
|
|
if query_object.sys_id:
|
|
filters.append(DataStdMain.sys_id == query_object.sys_id)
|
|
if query_object.std_type:
|
|
filters.append(DataStdMain.std_type == query_object.std_type)
|
|
if query_object.std_status:
|
|
filters.append(DataStdMain.std_status == query_object.std_status)
|
|
if query_object.content_onum:
|
|
filters.append(DataStdMain.content_onum == query_object.content_onum)
|
|
|
|
query = select(DataStdMain).where(*filters).order_by(desc(DataStdMain.create_time))
|
|
return await PageUtil.paginate(db, query, query_object.page_num, query_object.page_size, is_page)
|
|
@classmethod
|
|
async def get_std_main_list_all(cls, db: AsyncSession, query_object: DataStdMainModel):
|
|
filters = []
|
|
|
|
if query_object.std_name:
|
|
filters.append(DataStdMain.std_name.like(f"%{query_object.std_name}%"))
|
|
if query_object.std_code:
|
|
filters.append(DataStdMain.std_code.like(f"%{query_object.std_code}%"))
|
|
if query_object.sys_id:
|
|
filters.append(DataStdMain.sys_id == query_object.sys_id)
|
|
if query_object.code_id:
|
|
filters.append(DataStdMain.code_id == query_object.code_id)
|
|
if query_object.std_type:
|
|
filters.append(DataStdMain.std_type == query_object.std_type)
|
|
if query_object.std_status:
|
|
filters.append(DataStdMain.std_status == query_object.std_status)
|
|
query = select(DataStdMain).where(*filters).order_by(desc(DataStdMain.create_time))
|
|
return await PageUtil.paginate(db, query, 0, 0, False)
|
|
@classmethod
|
|
async def add_std_main(cls, db: AsyncSession, model: DataStdMainModel):
|
|
col = DataStdMain(**model.model_dump())
|
|
db.add(col)
|
|
await db.flush()
|
|
return col
|
|
|
|
@classmethod
|
|
async def delete_std_main(cls, db: AsyncSession, Id: str):
|
|
await db.execute(delete(DataStdMain).where(DataStdMain.id == Id))
|
|
|
|
@classmethod
|
|
async def update_std_main(cls, db: AsyncSession, update_data: DataStdMainModel):
|
|
await db.execute(update(DataStdMain), [update_data])
|
|
await db.flush()
|
|
|
|
@classmethod
|
|
async def get_std_main_by_id(cls, db: AsyncSession, Id: str):
|
|
col = await db.execute(select(DataStdMain).where(DataStdMain.id == Id))
|
|
return col.scalars().first()
|
|
|
|
@classmethod
|
|
async def get_data_main_by_info(cls, db: AsyncSession, query_object: DataStdMainModel):
|
|
col = await db.execute(
|
|
select(DataStdMain).where(
|
|
DataStdMain.std_name == query_object.std_name if query_object.std_name else True,
|
|
DataStdMain.std_code == query_object.std_code if query_object.std_code else True,
|
|
DataStdMain.std_num == query_object.std_num if query_object.std_num else True,
|
|
DataStdMain.sys_id == query_object.sys_id if query_object.sys_id else True,
|
|
DataStdMain.std_type == query_object.std_type if query_object.std_type else True,
|
|
DataStdMain.code_id == query_object.code_id if query_object.code_id else True,
|
|
DataStdMain.std_status == query_object.std_status if query_object.std_status else True
|
|
)
|
|
)
|
|
return col.scalars().first()
|
|
|