86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
|
from sqlalchemy import select, update, delete, func
|
||
|
|
from typing import List, Optional
|
||
|
|
from module_admin.system.entity.do.etest_entities_do import TestPermission
|
||
|
|
|
||
|
|
|
||
|
|
class TestPermissionDao:
|
||
|
|
"""
|
||
|
|
测试权限配置数据访问层
|
||
|
|
"""
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def get_by_id(cls, db: AsyncSession, id: int) -> Optional[TestPermission]:
|
||
|
|
"""根据ID查询"""
|
||
|
|
result = await db.execute(
|
||
|
|
select(TestPermission).where(TestPermission.id == id, TestPermission.del_flag == '0')
|
||
|
|
)
|
||
|
|
return result.scalar_one_or_none()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def get_by_user_id(cls, db: AsyncSession, user_id: int) -> List[TestPermission]:
|
||
|
|
"""根据用户ID查询权限列表"""
|
||
|
|
result = await db.execute(
|
||
|
|
select(TestPermission).where(
|
||
|
|
TestPermission.user_id == user_id,
|
||
|
|
TestPermission.del_flag == '0',
|
||
|
|
TestPermission.can_claim == True
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return result.scalars().all()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def get_by_user_and_category(cls, db: AsyncSession, user_id: int, category_id: int) -> Optional[TestPermission]:
|
||
|
|
"""根据用户ID和类别ID查询权限"""
|
||
|
|
result = await db.execute(
|
||
|
|
select(TestPermission).where(
|
||
|
|
TestPermission.user_id == user_id,
|
||
|
|
TestPermission.test_category_id == category_id,
|
||
|
|
TestPermission.del_flag == '0',
|
||
|
|
TestPermission.can_claim == True
|
||
|
|
)
|
||
|
|
)
|
||
|
|
return result.scalar_one_or_none()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def get_list(cls, db: AsyncSession, query_obj, is_page: bool = False):
|
||
|
|
"""查询列表"""
|
||
|
|
query = select(TestPermission).where(TestPermission.del_flag == '0')
|
||
|
|
|
||
|
|
if query_obj.user_id:
|
||
|
|
query = query.where(TestPermission.user_id == query_obj.user_id)
|
||
|
|
if query_obj.test_category_id:
|
||
|
|
query = query.where(TestPermission.test_category_id == query_obj.test_category_id)
|
||
|
|
if query_obj.permission_type:
|
||
|
|
query = query.where(TestPermission.permission_type == query_obj.permission_type)
|
||
|
|
|
||
|
|
if is_page:
|
||
|
|
total = await db.execute(select(func.count()).select_from(query.subquery()))
|
||
|
|
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.scalar()
|
||
|
|
else:
|
||
|
|
result = await db.execute(query)
|
||
|
|
return result.scalars().all()
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def add(cls, db: AsyncSession, obj: TestPermission):
|
||
|
|
"""新增"""
|
||
|
|
db.add(obj)
|
||
|
|
await db.flush()
|
||
|
|
return obj.id
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def update(cls, db: AsyncSession, obj):
|
||
|
|
"""更新"""
|
||
|
|
await db.execute(
|
||
|
|
update(TestPermission).where(TestPermission.id == obj.id).values(**obj)
|
||
|
|
)
|
||
|
|
|
||
|
|
@classmethod
|
||
|
|
async def delete(cls, db: AsyncSession, ids: List[int]):
|
||
|
|
"""删除（逻辑删除）"""
|
||
|
|
await db.execute(
|
||
|
|
update(TestPermission).where(TestPermission.id.in_(ids)).values(del_flag='1')
|
||
|
|
)
|