65 lines
2.7 KiB
Python
65 lines
2.7 KiB
Python
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from module_business.entity.do.master_do import BizUserSite, BizUserWarehouse
|
|
|
|
|
|
class AccessDao:
|
|
"""仓库与客户现场数据权限数据库操作层。"""
|
|
|
|
@classmethod
|
|
async def user_has_warehouse(cls, db: AsyncSession, user_id: int, warehouse_id: int) -> bool:
|
|
"""判断用户是否关联指定仓库。"""
|
|
mapped_user_id = (
|
|
await db.execute(
|
|
select(BizUserWarehouse.user_id).where(
|
|
BizUserWarehouse.user_id == user_id,
|
|
BizUserWarehouse.warehouse_id == warehouse_id,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
return mapped_user_id is not None
|
|
|
|
@classmethod
|
|
async def user_has_site(cls, db: AsyncSession, user_id: int, site_id: int) -> bool:
|
|
"""判断用户是否关联指定客户现场。"""
|
|
mapped_user_id = (
|
|
await db.execute(
|
|
select(BizUserSite.user_id).where(
|
|
BizUserSite.user_id == user_id,
|
|
BizUserSite.site_id == site_id,
|
|
)
|
|
)
|
|
).scalar_one_or_none()
|
|
|
|
return mapped_user_id is not None
|
|
|
|
@classmethod
|
|
async def list_user_warehouse_ids(cls, db: AsyncSession, user_id: int) -> list[int]:
|
|
"""获取用户关联的仓库 ID 列表。"""
|
|
result = await db.execute(select(BizUserWarehouse.warehouse_id).where(BizUserWarehouse.user_id == user_id))
|
|
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def list_user_site_ids(cls, db: AsyncSession, user_id: int) -> list[int]:
|
|
"""获取用户关联的客户现场 ID 列表。"""
|
|
result = await db.execute(select(BizUserSite.site_id).where(BizUserSite.user_id == user_id))
|
|
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def replace_warehouse_users(cls, db: AsyncSession, warehouse_id: int, user_ids: list[int]) -> None:
|
|
"""完整替换指定仓库的用户关联，仅刷新当前事务。"""
|
|
await db.execute(delete(BizUserWarehouse).where(BizUserWarehouse.warehouse_id == warehouse_id))
|
|
db.add_all([BizUserWarehouse(user_id=user_id, warehouse_id=warehouse_id) for user_id in user_ids])
|
|
await db.flush()
|
|
|
|
@classmethod
|
|
async def replace_site_users(cls, db: AsyncSession, site_id: int, user_ids: list[int]) -> None:
|
|
"""完整替换指定客户现场的用户关联，仅刷新当前事务。"""
|
|
await db.execute(delete(BizUserSite).where(BizUserSite.site_id == site_id))
|
|
db.add_all([BizUserSite(user_id=user_id, site_id=site_id) for user_id in user_ids])
|
|
await db.flush()
|