Browse Source

feat: Add universal search for devices

Backend changes:
- Added search parameter to device_service.list_devices()
- Search across: MAC address, simple_id, organization name/email
- Minimum 2 characters for search
- Case-insensitive (ILIKE) search with LEFT JOIN to organizations
- Updated superadmin and client endpoints with search parameter

Features:
- Universal search box as in legacy admin panel
- Searches all relevant fields in one query
- Efficient SQL with LEFT OUTER JOIN

Next: Frontend search input with debounce

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
root 1 month ago
parent
commit
4a4a2cc36a

+ 10 - 0
backend/app/api/v1/client/devices.py

@@ -23,11 +23,20 @@ async def list_organization_devices(
     skip: int = Query(0, ge=0, description="Number of records to skip"),
     limit: int = Query(100, ge=1, le=1000, description="Max records to return"),
     status: str | None = Query(None, description="Filter by status"),
+    search: str | None = Query(
+        None,
+        min_length=2,
+        description="Universal search: MAC, simple_id",
+    ),
 ):
     """
     List devices assigned to current user's organization.
 
     All authenticated users can view devices in their organization.
+
+    Search parameter searches across:
+    - MAC address
+    - Simple ID (#1, #2, etc)
     """
     if not current_user.organization_id:
         raise HTTPException(
@@ -41,6 +50,7 @@ async def list_organization_devices(
         limit=limit,
         organization_id=current_user.organization_id,
         status=status,
+        search=search,
     )
 
     return DeviceListResponse(

+ 12 - 0
backend/app/api/v1/superadmin/devices.py

@@ -31,11 +31,22 @@ async def list_devices(
         None, description="Filter by organization"
     ),
     status: str | None = Query(None, description="Filter by status"),
+    search: str | None = Query(
+        None,
+        min_length=2,
+        description="Universal search: MAC, simple_id, organization name/email",
+    ),
 ):
     """
     List all devices (superadmin only).
 
     Returns paginated list of devices with optional filters.
+
+    Search parameter searches across:
+    - MAC address
+    - Simple ID (#1, #2, etc)
+    - Organization name
+    - Organization contact email
     """
     devices, total = await device_service.list_devices(
         db,
@@ -43,6 +54,7 @@ async def list_devices(
         limit=limit,
         organization_id=organization_id,
         status=status,
+        search=search,
     )
 
     return DeviceListResponse(

+ 43 - 10
backend/app/services/device_service.py

@@ -4,10 +4,12 @@ Device management service.
 
 from datetime import datetime, timezone
 
-from sqlalchemy import func, select
+from sqlalchemy import String, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import joinedload
 
 from app.models.device import Device
+from app.models.organization import Organization
 from app.schemas.device import DeviceCreate, DeviceUpdate
 
 
@@ -87,6 +89,7 @@ async def list_devices(
     limit: int = 100,
     organization_id: int | None = None,
     status: str | None = None,
+    search: str | None = None,
 ) -> tuple[list[Device], int]:
     """
     List devices with pagination and filters.
@@ -97,33 +100,63 @@ async def list_devices(
         limit: Maximum number of records to return
         organization_id: Filter by organization (optional)
         status: Filter by status (optional)
+        search: Universal search across all fields (optional)
 
     Returns:
         Tuple of (devices list, total count)
     """
-    # Build query
-    query = select(Device)
+    # Build query with Organization join for search
+    query = select(Device).options(joinedload(Device.organization))
+
+    # Base filters
+    filters = []
 
     if organization_id is not None:
-        query = query.where(Device.organization_id == organization_id)
+        filters.append(Device.organization_id == organization_id)
 
     if status:
-        query = query.where(Device.status == status)
+        filters.append(Device.status == status)
+
+    # Universal search filter
+    if search and len(search) >= 2:
+        # Join with Organization for searching by org name/email
+        query = query.join(Organization, Device.organization_id == Organization.id, isouter=True)
+
+        # Search across multiple fields
+        search_pattern = f"%{search}%"
+        search_filters = [
+            Device.mac_address.ilike(search_pattern),
+            func.cast(Device.simple_id, String).ilike(search_pattern),
+        ]
+
+        # Only add organization filters if we have organization data
+        search_filters.extend([
+            Organization.name.ilike(search_pattern),
+            Organization.contact_email.ilike(search_pattern),
+        ])
+
+        filters.append(or_(*search_filters))
+
+    # Apply all filters
+    if filters:
+        query = query.where(*filters)
 
     # Get total count
     count_query = select(func.count()).select_from(Device)
 
-    if organization_id is not None:
-        count_query = count_query.where(Device.organization_id == organization_id)
+    if search and len(search) >= 2:
+        count_query = count_query.join(
+            Organization, Device.organization_id == Organization.id, isouter=True
+        )
 
-    if status:
-        count_query = count_query.where(Device.status == status)
+    if filters:
+        count_query = count_query.where(*filters)
 
     total_result = await db.execute(count_query)
     total = total_result.scalar_one()
 
     # Get paginated results
-    query = query.offset(skip).limit(limit).order_by(Device.simple_id)
+    query = query.offset(skip).limit(limit).order_by(Device.simple_id.desc())
     result = await db.execute(query)
     devices = list(result.scalars().all())