106 lines
3.9 KiB
Python
106 lines
3.9 KiB
Python
from sqlalchemy import delete, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from module_business.domain.test_batch_scope import TestBatchScope, batch_scope_condition
|
|
from module_business.entity.do.master_do import BizCustomerSite, BizUserSite, BizUserWarehouse, BizWarehouse
|
|
|
|
|
|
class AccessDao:
|
|
"""仓库与客户现场数据权限数据库操作层。"""
|
|
|
|
@classmethod
|
|
async def user_has_warehouse(
|
|
cls,
|
|
db: AsyncSession,
|
|
user_id: int,
|
|
warehouse_id: int,
|
|
*,
|
|
for_update: bool = False,
|
|
) -> bool:
|
|
"""判断用户是否关联指定仓库。"""
|
|
statement = select(BizUserWarehouse.user_id).where(
|
|
BizUserWarehouse.user_id == user_id,
|
|
BizUserWarehouse.warehouse_id == warehouse_id,
|
|
)
|
|
if for_update:
|
|
statement = statement.with_for_update()
|
|
mapped_user_id = (await db.execute(statement)).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,
|
|
*,
|
|
for_update: bool = False,
|
|
) -> bool:
|
|
"""判断用户是否关联指定客户现场。"""
|
|
statement = select(BizUserSite.user_id).where(
|
|
BizUserSite.user_id == user_id,
|
|
BizUserSite.site_id == site_id,
|
|
)
|
|
if for_update:
|
|
statement = statement.with_for_update()
|
|
mapped_user_id = (await db.execute(statement)).scalar_one_or_none()
|
|
|
|
return mapped_user_id is not None
|
|
|
|
@classmethod
|
|
async def list_user_warehouse_ids(
|
|
cls,
|
|
db: AsyncSession,
|
|
user_id: int,
|
|
scope: TestBatchScope | None = None,
|
|
) -> list[int]:
|
|
"""获取用户关联且处于当前批次作用域的仓库 ID 列表。"""
|
|
conditions = [BizUserWarehouse.user_id == user_id]
|
|
if scope is not None:
|
|
conditions.append(batch_scope_condition(BizWarehouse.test_batch_id, scope))
|
|
statement = (
|
|
select(BizUserWarehouse.warehouse_id)
|
|
.select_from(BizUserWarehouse)
|
|
.join(BizWarehouse, BizWarehouse.warehouse_id == BizUserWarehouse.warehouse_id)
|
|
.where(*conditions)
|
|
)
|
|
result = await db.execute(statement)
|
|
|
|
return list(result.scalars().all())
|
|
|
|
@classmethod
|
|
async def list_user_site_ids(
|
|
cls,
|
|
db: AsyncSession,
|
|
user_id: int,
|
|
scope: TestBatchScope | None = None,
|
|
) -> list[int]:
|
|
"""获取用户关联且处于当前批次作用域的客户现场 ID 列表。"""
|
|
conditions = [BizUserSite.user_id == user_id]
|
|
if scope is not None:
|
|
conditions.append(batch_scope_condition(BizCustomerSite.test_batch_id, scope))
|
|
statement = (
|
|
select(BizUserSite.site_id)
|
|
.select_from(BizUserSite)
|
|
.join(BizCustomerSite, BizCustomerSite.site_id == BizUserSite.site_id)
|
|
.where(*conditions)
|
|
)
|
|
result = await db.execute(statement)
|
|
|
|
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()
|