114 lines
3.4 KiB
Python
114 lines
3.4 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy import update as sa_update
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.permission import Permission
|
|
from app.models.role import Role
|
|
|
|
|
|
async def get(db: AsyncSession, role_id: int) -> Role | None:
|
|
return await db.get(Role, role_id)
|
|
|
|
|
|
async def get_by_name(db: AsyncSession, name: str) -> Role | None:
|
|
result = await db.execute(select(Role).where(Role.name == name))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def get_default(db: AsyncSession) -> Role | None:
|
|
"""The role auto-assigned to new users, or ``None`` if none is marked."""
|
|
result = await db.execute(select(Role).where(Role.is_default.is_(True)))
|
|
return result.scalars().first()
|
|
|
|
|
|
async def set_default(db: AsyncSession, role: Role) -> Role:
|
|
"""Make ``role`` the sole default, clearing the flag on every other role."""
|
|
await db.execute(sa_update(Role).where(Role.id != role.id).values(is_default=False))
|
|
role.is_default = True
|
|
await db.commit()
|
|
await db.refresh(role)
|
|
return role
|
|
|
|
|
|
async def list_roles(db: AsyncSession) -> list[Role]:
|
|
result = await db.execute(select(Role).order_by(Role.id))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def _resolve_permissions(db: AsyncSession, ids: list[int]) -> list[Permission]:
|
|
if not ids:
|
|
return []
|
|
result = await db.execute(select(Permission).where(Permission.id.in_(ids)))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
name: str,
|
|
description: str | None = None,
|
|
permission_ids: list[int] | None = None,
|
|
) -> Role:
|
|
role = Role(
|
|
name=name,
|
|
description=description,
|
|
permissions=await _resolve_permissions(db, permission_ids or []),
|
|
)
|
|
db.add(role)
|
|
await db.commit()
|
|
await db.refresh(role)
|
|
return role
|
|
|
|
|
|
async def update(db: AsyncSession, role: Role, data: dict) -> Role:
|
|
if "permission_ids" in data and data["permission_ids"] is not None:
|
|
role.permissions = await _resolve_permissions(db, data.pop("permission_ids"))
|
|
else:
|
|
data.pop("permission_ids", None)
|
|
|
|
# Setting a role default must clear the flag on all others.
|
|
make_default = data.pop("is_default", None)
|
|
if make_default:
|
|
await db.execute(
|
|
sa_update(Role).where(Role.id != role.id).values(is_default=False)
|
|
)
|
|
role.is_default = True
|
|
elif make_default is False:
|
|
role.is_default = False
|
|
|
|
for field, value in data.items():
|
|
if value is not None:
|
|
setattr(role, field, value)
|
|
|
|
await db.commit()
|
|
await db.refresh(role)
|
|
return role
|
|
|
|
|
|
async def delete(db: AsyncSession, role: Role) -> None:
|
|
await db.delete(role)
|
|
await db.commit()
|
|
|
|
|
|
# --- Permissions -------------------------------------------------------------
|
|
|
|
|
|
async def list_permissions(db: AsyncSession) -> list[Permission]:
|
|
result = await db.execute(select(Permission).order_by(Permission.name))
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def get_permission_by_name(db: AsyncSession, name: str) -> Permission | None:
|
|
result = await db.execute(select(Permission).where(Permission.name == name))
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def create_permission(
|
|
db: AsyncSession, *, name: str, description: str | None = None
|
|
) -> Permission:
|
|
perm = Permission(name=name, description=description)
|
|
db.add(perm)
|
|
await db.commit()
|
|
await db.refresh(perm)
|
|
return perm
|