sprint3: dashboard counts per user + import owner_id + export visibility filter

app/routes/dashboard.py

@@ -2,11 +2,16 @@
 
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends
+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"])
@@ -36,3 +41,60 @@ async def list_dashboard_widgets(
     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,
+    }

app/routes/import_export.py

@@ -92,15 +92,21 @@ async def export_data(
       - format: 'csv' or 'xlsx' (default: csv)
     """
     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)
 
     # Determine filename based on entity_type
     filename = f"{entity_type}_export.csv"
 
     # Fetch CSV data from the appropriate service function
     if entity_type == "contacts":
-        csv_data = await import_export_service.export_contacts_csv(db, tenant_id)
+        csv_data = await import_export_service.export_contacts_csv(
+            db, tenant_id, user_id=user_id, is_system_admin=is_system_admin
+        )
     elif entity_type == "companies":
-        csv_data = await import_export_service.export_companies_csv(db, tenant_id)
+        csv_data = await import_export_service.export_companies_csv(
+            db, tenant_id, user_id=user_id, is_system_admin=is_system_admin
+        )
     else:
         raise HTTPException(status_code=400, detail=f"Unsupported entity_type: {entity_type}")
 

app/services/import_export_service.py

@@ -15,6 +15,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.core.audit import log_audit
+from app.core.visibility import apply_visibility_filter
 from app.models.contact import Contact
 from app.services.contact_service import _serialize_contact as _contact_to_dict
 
@@ -86,6 +87,7 @@ async def import_companies(
             phone_1=row.get("phone", "").strip() or None,
             website=row.get("website", "").strip() or None,
             description=row.get("description", "").strip() or None,
+            owner_id=user_id,
             created_by=user_id,
             updated_by=user_id,
         )
@@ -162,6 +164,7 @@ async def import_contacts(
             email_1=row.get("email", "").strip() or None,
             phone_1=row.get("phone", "").strip() or None,
             phone_2=row.get("mobile", "").strip() or None,
+            owner_id=user_id,
             created_by=user_id,
             updated_by=user_id,
         )
@@ -215,6 +218,8 @@ async def import_csv(
 async def export_contacts_csv(
     db: AsyncSession,
     tenant_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> str:
     """Export contacts as CSV string using unified Contact model fields."""
     q = (
@@ -225,6 +230,10 @@ async def export_contacts_csv(
         )
         .order_by(Contact.surname, Contact.firstname)
     )
+    if user_id:
+        q = await apply_visibility_filter(
+            db, q, "contact", Contact, user_id, tenant_id, is_system_admin
+        )
     result = await db.execute(q)
     contacts = result.scalars().all()
 
@@ -255,6 +264,8 @@ async def export_contacts_csv(
 async def export_companies_csv(
     db: AsyncSession,
     tenant_id: uuid.UUID,
+    user_id: uuid.UUID | None = None,
+    is_system_admin: bool = False,
 ) -> str:
     """Export companies (Contact.type == 'company') as CSV string."""
     q = (
@@ -266,6 +277,10 @@ async def export_companies_csv(
         )
         .order_by(Contact.name)
     )
+    if user_id:
+        q = await apply_visibility_filter(
+            db, q, "contact", Contact, user_id, tenant_id, is_system_admin
+        )
     result = await db.execute(q)
     companies = result.scalars().all()