finalized search backend

backend/api/search.py

@@ -1,9 +1,8 @@
 from fastapi import APIRouter, Depends
 
-from ..services import SearchService
+from backend.services.search import SearchResult
-from ..models.resource_model import Resource
-from ..models.service_model import Service
 
+from ..services import SearchService
 
 api = APIRouter(prefix="/api/search")
 
@@ -12,7 +11,7 @@ openapi_tags = {
     "description": "Search through all resources and services for a string.",
 }
 
-@api.post("", tags=["Search"])
+@api.get("", tags=["Search"])
-def search(query: str, search_svc: SearchService = Depends()) -> list[Resource | Service]:
+def search(query: str, search_svc: SearchService = Depends()) -> list[SearchResult]:
     return search_svc.search(query)
   

backend/services/search.py

@@ -1,37 +1,72 @@
+from typing import Literal
 from fastapi import Depends
+from pydantic import BaseModel
+
+from backend.entities.user_entity import UserEntity
+from backend.models.user_model import User
+
 from ..database import db_session
 from sqlalchemy.orm import Session
-from sqlalchemy import or_, select
+from sqlalchemy import (
+    ARRAY,
+    BinaryExpression,
+    String,
+    Text,
+    literal_column,
+    or_,
+    func,
+    select,
+    exists,
+)
 from ..models.resource_model import Resource
 from ..models.service_model import Service
 from ..entities.resource_entity import ResourceEntity
 from backend.entities.service_entity import ServiceEntity
-from ..models.user_model import User, UserTypeEnum
+
+
+class SearchResult(BaseModel):
+    type: Literal["resource", "service", "user"]
+    data: Resource | Service | User
 
 
 class SearchService:
     def __init__(self, session: Session = Depends(db_session)):
         self._session = session
 
-    def search(self, query_str: str) -> list[Resource | Service]:
+    def search(self, query_str: str) -> list[SearchResult]:
-        results = []
+        """Searches through all tables for a string."""
-        models = [
+        results: list[SearchResult] = []
+        entities = (
             ResourceEntity,
             ServiceEntity,
-        ]
+            UserEntity,
+        )
 
-        for model in models:
+        for entity in entities:
-            columns = [column for column in model.__table__.columns]
+            columns = entity.__table__.columns
 
-            filters = [
+            filters: list[BinaryExpression[bool]] = []
-                column.ilike(f"%{query_str}%")
+            for column in columns:
-                for column in columns
+                if isinstance(column.type, String) or isinstance(column.type, Text):
-                if column.type.__class__.__name__ in ["String", "Text"]
+                    filters.append(column.cast(String).ilike(f"%{query_str}%"))
-            ]
+                elif isinstance(column.type, ARRAY):
+                    # Custom filter that checks the query string against each element in the array
+                    filters.append(
+                        exists(
+                            select(1)
+                            .select_from(
+                                func.unnest(column.cast(ARRAY(String))).alias("element")
+                            )
+                            .where(literal_column("element").ilike(f"%{query_str}%"))
+                        )
+                    )
 
             if filters:
-                query = self._session.query(model).filter(or_(*filters))
+                query = self._session.query(entity).filter(or_(*filters))
-                results.extend(query.all())
+                results.extend(
-
+                    [
+                        SearchResult(type=entity.__tablename__, data=result.to_model())
+                        for result in query.all()
+                    ]
+                )
         return results
-