Agent Zero
101 lines
3.5 KiB
Python
101 lines
3.5 KiB
Python
"""Dashboard routes — list available widgets from active plugins (Task 5.25)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.db import get_db
|
|
from app.core.visibility import apply_visibility_filter
|
|
from app.deps import require_permission
|
|
from app.models.contact import Contact
|
|
from app.plugins.registry import get_registry
|
|
|
|
router = APIRouter(prefix="/api/v1/dashboard", tags=["dashboard"])
|
|
|
|
|
|
@router.get("/widgets")
|
|
async def list_dashboard_widgets(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(require_permission("dashboard:read")),
|
|
):
|
|
"""List all available dashboard widgets from active plugins.
|
|
|
|
Returns a flat list of widget definitions contributed by active plugins,
|
|
sorted by their order field. Each widget includes the contributing plugin name.
|
|
"""
|
|
registry = get_registry()
|
|
manifests = await registry.get_active_manifests(db)
|
|
|
|
widgets: list[dict] = []
|
|
for manifest in manifests:
|
|
for widget in manifest.get("dashboard_widgets", []):
|
|
widget_copy = dict(widget)
|
|
widget_copy["plugin_name"] = manifest["name"]
|
|
widgets.append(widget_copy)
|
|
|
|
# Sort by order field
|
|
widgets.sort(key=lambda w: w.get("order", 100))
|
|
|
|
return {"items": widgets, "total": len(widgets)}
|
|
|
|
|
|
@router.get("/counts")
|
|
async def get_dashboard_counts(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: dict = Depends(require_permission("dashboard:read")),
|
|
):
|
|
"""Get dashboard count statistics filtered by user visibility.
|
|
|
|
Returns counts for contacts and companies that the current user
|
|
is allowed to see based on ownership and sharing permissions.
|
|
"""
|
|
tenant_id = uuid.UUID(current_user["tenant_id"])
|
|
user_id = uuid.UUID(current_user["user_id"])
|
|
is_system_admin = current_user.get("is_system_admin", False)
|
|
|
|
# Contact count with visibility filter
|
|
contact_query = select(func.count(Contact.id)).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.deleted_at.is_(None),
|
|
)
|
|
contact_query = await apply_visibility_filter(
|
|
db, contact_query, "contact", Contact, user_id, tenant_id, is_system_admin
|
|
)
|
|
contact_result = await db.execute(contact_query)
|
|
contact_count = contact_result.scalar() or 0
|
|
|
|
# Company count (Contact.type == 'company') with visibility filter
|
|
company_query = select(func.count(Contact.id)).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.deleted_at.is_(None),
|
|
Contact.type == "company",
|
|
)
|
|
company_query = await apply_visibility_filter(
|
|
db, company_query, "contact", Contact, user_id, tenant_id, is_system_admin
|
|
)
|
|
company_result = await db.execute(company_query)
|
|
company_count = company_result.scalar() or 0
|
|
|
|
# Person count (Contact.type == 'person') with visibility filter
|
|
person_query = select(func.count(Contact.id)).where(
|
|
Contact.tenant_id == tenant_id,
|
|
Contact.deleted_at.is_(None),
|
|
Contact.type == "person",
|
|
)
|
|
person_query = await apply_visibility_filter(
|
|
db, person_query, "contact", Contact, user_id, tenant_id, is_system_admin
|
|
)
|
|
person_result = await db.execute(person_query)
|
|
person_count = person_result.scalar() or 0
|
|
|
|
return {
|
|
"contacts": contact_count,
|
|
"companies": company_count,
|
|
"persons": person_count,
|
|
"total": contact_count,
|
|
}
|