leocrm/app/services/saved_view_service.py

"""SavedView service — CRUD with tenant isolation and visibility filter."""

from __future__ import annotations

import uuid
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.visibility import apply_visibility_filter, check_single_entity_access
from app.models.saved_view import SavedView


def _view_to_dict(v: SavedView) -> dict[str, Any]:
    """Serialize a SavedView ORM object to dict."""
    return {
        "id": str(v.id),
        "name": v.name,
        "entity_type": v.entity_type,
        "view_config": v.view_config,
        "user_id": str(v.user_id),
        "created_at": v.created_at.isoformat() if v.created_at else None,
        "updated_at": v.updated_at.isoformat() if v.updated_at else None,
    }


async def list_saved_views(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    entity_type: str | None = None,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> list[dict[str, Any]]:
    """List saved views for a tenant, optionally filtered by entity_type."""
    q = select(SavedView).where(
        SavedView.tenant_id == tenant_id,
        SavedView.deleted_at.is_(None),
    )
    if entity_type:
        q = q.where(SavedView.entity_type == entity_type)
    if user_id and not is_system_admin:
        q = await apply_visibility_filter(
            db, q, "saved_view", SavedView, user_id, tenant_id, is_system_admin
        )
    q = q.order_by(SavedView.name)
    result = await db.execute(q)
    return [_view_to_dict(v) for v in result.scalars().all()]


async def get_saved_view(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    view_id: uuid.UUID,
    user_id: uuid.UUID | None = None,
    is_system_admin: bool = False,
) -> dict[str, Any] | None:
    """Get a single saved view by ID."""
    q = select(SavedView).where(
        SavedView.id == view_id,
        SavedView.tenant_id == tenant_id,
        SavedView.deleted_at.is_(None),
    )
    result = await db.execute(q)
    saved = result.scalar_one_or_none()
    if saved is None:
        return None
    if user_id and not is_system_admin:
        has_access = await check_single_entity_access(
            db, "saved_view", saved.id, user_id, tenant_id, "read", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")
    return _view_to_dict(saved)


async def create_saved_view(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    data: dict[str, Any],
) -> dict[str, Any]:
    """Create a new saved view."""
    saved = SavedView(
        tenant_id=tenant_id,
        user_id=user_id,
        name=data["name"],
        entity_type=data["entity_type"],
        view_config=data.get("view_config", {}),
        owner_id=user_id,
    )
    db.add(saved)
    await db.flush()
    await db.refresh(saved)
    return _view_to_dict(saved)


async def update_saved_view(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    view_id: uuid.UUID,
    data: dict[str, Any],
    is_system_admin: bool = False,
) -> dict[str, Any] | None:
    """Update a saved view."""
    q = select(SavedView).where(
        SavedView.id == view_id,
        SavedView.tenant_id == tenant_id,
        SavedView.deleted_at.is_(None),
    )
    result = await db.execute(q)
    saved = result.scalar_one_or_none()
    if saved is None:
        return None

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "saved_view", saved.id, user_id, tenant_id, "write", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    if "name" in data and data["name"] is not None:
        saved.name = data["name"]
    if "view_config" in data and data["view_config"] is not None:
        saved.view_config = data["view_config"]

    await db.flush()
    await db.refresh(saved)
    return _view_to_dict(saved)


async def delete_saved_view(
    db: AsyncSession,
    tenant_id: uuid.UUID,
    user_id: uuid.UUID,
    view_id: uuid.UUID,
    is_system_admin: bool = False,
) -> bool:
    """Soft-delete a saved view."""
    q = select(SavedView).where(
        SavedView.id == view_id,
        SavedView.tenant_id == tenant_id,
        SavedView.deleted_at.is_(None),
    )
    result = await db.execute(q)
    saved = result.scalar_one_or_none()
    if saved is None:
        return False

    if not is_system_admin:
        has_access = await check_single_entity_access(
            db, "saved_view", saved.id, user_id, tenant_id, "admin", is_system_admin
        )
        if not has_access:
            raise PermissionError("No access")

    saved.deleted_at = datetime.now(UTC)
    await db.flush()
    return True