81 lines
2.7 KiB
Python
81 lines
2.7 KiB
Python
"""
|
|
测试表单模板表数据访问层
|
|
"""
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, update, func, desc
|
|
from typing import List, Optional
|
|
from module_admin.system.entity.do.test_form_do import TestForm
|
|
|
|
|
|
class TestFormDao:
|
|
"""
|
|
测试表单模板表数据访问层
|
|
"""
|
|
|
|
@classmethod
|
|
async def get_by_id(cls, db: AsyncSession, id: int) -> Optional[TestForm]:
|
|
"""根据ID查询"""
|
|
result = await db.execute(
|
|
select(TestForm).where(TestForm.id == id, TestForm.del_flag == '0')
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
@classmethod
|
|
async def get_by_type(cls, db: AsyncSession, form_type: str) -> List[TestForm]:
|
|
"""根据类型查询"""
|
|
result = await db.execute(
|
|
select(TestForm).where(
|
|
TestForm.form_type == form_type,
|
|
TestForm.del_flag == '0',
|
|
TestForm.is_active == 1
|
|
).order_by(desc(TestForm.create_time))
|
|
)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def get_list(cls, db: AsyncSession, query_obj, is_page: bool = True):
|
|
"""查询列表"""
|
|
query = select(TestForm).where(TestForm.del_flag == '0')
|
|
|
|
if query_obj.name:
|
|
query = query.where(TestForm.name.like(f'%{query_obj.name}%'))
|
|
if query_obj.form_type:
|
|
query = query.where(TestForm.form_type == query_obj.form_type)
|
|
if query_obj.is_active is not None:
|
|
query = query.where(TestForm.is_active == query_obj.is_active)
|
|
|
|
query = query.order_by(desc(TestForm.create_time))
|
|
|
|
if is_page:
|
|
# 分页
|
|
total_result = await db.execute(select(func.count()).select_from(query.subquery()))
|
|
total = total_result.scalar()
|
|
|
|
query = query.offset((query_obj.page_num - 1) * query_obj.page_size).limit(query_obj.page_size)
|
|
result = await db.execute(query)
|
|
return result.scalars().all(), total
|
|
else:
|
|
result = await db.execute(query)
|
|
return result.scalars().all()
|
|
|
|
@classmethod
|
|
async def add(cls, db: AsyncSession, obj: TestForm):
|
|
"""新增"""
|
|
db.add(obj)
|
|
await db.flush()
|
|
return obj.id
|
|
|
|
@classmethod
|
|
async def update(cls, db: AsyncSession, obj: TestForm):
|
|
"""更新"""
|
|
await db.merge(obj)
|
|
await db.flush()
|
|
|
|
@classmethod
|
|
async def delete(cls, db: AsyncSession, ids: List[int]):
|
|
"""删除（逻辑删除）"""
|
|
await db.execute(
|
|
update(TestForm).where(TestForm.id.in_(ids)).values(del_flag='1')
|
|
)
|
|
await db.flush()
|