34 lines
1.3 KiB
Python
34 lines
1.3 KiB
Python
"""User and RBAC repository queries."""
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.entities import RolePermission, UserAccount
|
|
|
|
|
|
class UserRepository:
|
|
def __init__(self, session: AsyncSession) -> None:
|
|
self.session = session
|
|
|
|
async def get_by_id(self, user_id: int) -> UserAccount | None:
|
|
return await self.session.get(UserAccount, user_id)
|
|
|
|
async def get_by_username(self, username: str) -> UserAccount | None:
|
|
statement = select(UserAccount).where(
|
|
(UserAccount.emp_no == username) | (UserAccount.email == username)
|
|
)
|
|
return await self.session.scalar(statement)
|
|
|
|
async def get_by_dingtalk_id(self, dingtalk_id: str) -> UserAccount | None:
|
|
statement = select(UserAccount).where(UserAccount.dingtalk_id == dingtalk_id)
|
|
return await self.session.scalar(statement)
|
|
|
|
async def permissions_for_role(self, role: str, *, operate: bool | None = None) -> list[RolePermission]:
|
|
filters = [RolePermission.role == role]
|
|
if operate is True:
|
|
filters.append(RolePermission.can_operate.is_(True))
|
|
elif operate is False:
|
|
filters.append(RolePermission.can_view.is_(True))
|
|
statement = select(RolePermission).where(*filters)
|
|
return list(await self.session.scalars(statement))
|