Fascicoli+Tassonomia+permessi
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
"""
|
||||
Servizio CRUD per i preset di permessi (sottoruoli nominati).
|
||||
|
||||
Admin e supervisor possono creare, modificare ed eliminare preset per il proprio tenant.
|
||||
"""
|
||||
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.core.exceptions import ConflictError, ForbiddenError, NotFoundError
|
||||
from app.models.permission_preset import PermissionPreset
|
||||
from app.models.user import User
|
||||
from app.schemas.permission_preset import PermissionPresetCreate, PermissionPresetUpdate
|
||||
|
||||
|
||||
class PermissionPresetService:
|
||||
def __init__(self, db: AsyncSession) -> None:
|
||||
self.db = db
|
||||
|
||||
def _require_supervisor_or_admin(self, user: User) -> None:
|
||||
if not user.is_supervisor_or_admin:
|
||||
raise ForbiddenError("Solo amministratori e supervisori possono gestire i preset")
|
||||
|
||||
async def list_presets(self, tenant_id: uuid.UUID) -> list[PermissionPreset]:
|
||||
"""Ritorna tutti i preset del tenant ordinati per nome."""
|
||||
result = await self.db.execute(
|
||||
select(PermissionPreset)
|
||||
.where(PermissionPreset.tenant_id == tenant_id)
|
||||
.order_by(PermissionPreset.name)
|
||||
)
|
||||
return list(result.scalars().all())
|
||||
|
||||
async def get_preset(self, preset_id: uuid.UUID, tenant_id: uuid.UUID) -> PermissionPreset:
|
||||
"""Recupera un preset per ID verificando che appartenga al tenant."""
|
||||
preset = await self.db.get(PermissionPreset, preset_id)
|
||||
if not preset or preset.tenant_id != tenant_id:
|
||||
raise NotFoundError("preset")
|
||||
return preset
|
||||
|
||||
async def create_preset(
|
||||
self,
|
||||
tenant_id: uuid.UUID,
|
||||
data: PermissionPresetCreate,
|
||||
created_by: User,
|
||||
) -> PermissionPreset:
|
||||
"""Crea un nuovo preset. Il nome deve essere unico per tenant."""
|
||||
self._require_supervisor_or_admin(created_by)
|
||||
|
||||
# Verifica unicita' nome
|
||||
existing = await self.db.execute(
|
||||
select(PermissionPreset).where(
|
||||
PermissionPreset.tenant_id == tenant_id,
|
||||
PermissionPreset.name == data.name,
|
||||
)
|
||||
)
|
||||
if existing.scalar_one_or_none():
|
||||
raise ConflictError(f"Esiste gia' un preset con nome '{data.name}'")
|
||||
|
||||
preset = PermissionPreset(
|
||||
tenant_id=tenant_id,
|
||||
name=data.name,
|
||||
description=data.description,
|
||||
can_read=data.can_read,
|
||||
can_send=data.can_send,
|
||||
can_manage=data.can_manage,
|
||||
can_conserve=data.can_conserve,
|
||||
created_by=created_by.id,
|
||||
)
|
||||
self.db.add(preset)
|
||||
await self.db.flush()
|
||||
await self.db.refresh(preset)
|
||||
return preset
|
||||
|
||||
async def update_preset(
|
||||
self,
|
||||
preset_id: uuid.UUID,
|
||||
tenant_id: uuid.UUID,
|
||||
data: PermissionPresetUpdate,
|
||||
updated_by: User,
|
||||
) -> PermissionPreset:
|
||||
"""Aggiorna un preset esistente."""
|
||||
self._require_supervisor_or_admin(updated_by)
|
||||
|
||||
preset = await self.get_preset(preset_id, tenant_id)
|
||||
|
||||
# Verifica unicita' nome se cambia
|
||||
if data.name is not None and data.name != preset.name:
|
||||
existing = await self.db.execute(
|
||||
select(PermissionPreset).where(
|
||||
PermissionPreset.tenant_id == tenant_id,
|
||||
PermissionPreset.name == data.name,
|
||||
)
|
||||
)
|
||||
if existing.scalar_one_or_none():
|
||||
raise ConflictError(f"Esiste gia' un preset con nome '{data.name}'")
|
||||
preset.name = data.name
|
||||
|
||||
if data.description is not None:
|
||||
preset.description = data.description
|
||||
if data.can_read is not None:
|
||||
preset.can_read = data.can_read
|
||||
if data.can_send is not None:
|
||||
preset.can_send = data.can_send
|
||||
if data.can_manage is not None:
|
||||
preset.can_manage = data.can_manage
|
||||
if data.can_conserve is not None:
|
||||
preset.can_conserve = data.can_conserve
|
||||
|
||||
await self.db.flush()
|
||||
await self.db.refresh(preset)
|
||||
return preset
|
||||
|
||||
async def delete_preset(
|
||||
self,
|
||||
preset_id: uuid.UUID,
|
||||
tenant_id: uuid.UUID,
|
||||
deleted_by: User,
|
||||
) -> None:
|
||||
"""Elimina un preset."""
|
||||
self._require_supervisor_or_admin(deleted_by)
|
||||
preset = await self.get_preset(preset_id, tenant_id)
|
||||
await self.db.delete(preset)
|
||||
await self.db.flush()
|
||||
Reference in New Issue
Block a user