All Table Search Backend

backend/api/search.py

@@ -0,0 +1,17 @@
+from fastapi import APIRouter, Depends
+
+from backend.services.search import SearchResult
+
+from ..services import SearchService
+
+api = APIRouter(prefix="/api/search")
+
+openapi_tags = {
+    "name": "Search",
+    "description": "Search through all resources and services for a string.",
+}
+
+@api.get("", tags=["Search"])
+def search(query: str, search_svc: SearchService = Depends()) -> list[SearchResult]:
+    return search_svc.search(query)
+

backend/main.py

@@ -3,7 +3,7 @@
 from fastapi.middleware.gzip import GZipMiddleware
 
 
-from .api import user, health, service, resource, tag
+from .api import user, health, service, resource, tag, search
 
 description = """
 Welcome to the **COMPASS** RESTful Application Programming Interface.
@@ -24,7 +24,7 @@
 
 app.add_middleware(GZipMiddleware)
 
-feature_apis = [user, health, service, resource, tag]
+feature_apis = [user, health, service, resource, tag, search]
 
 for feature_api in feature_apis:
     app.include_router(feature_api.api)

backend/services/__init__.py

@@ -1,4 +1,5 @@
 from .user import UserService
 from .resource import ResourceService
 from .tag import TagService
-from .service import ServiceService
+from .service import ServiceService
+from .search import SearchService

backend/services/search.py

@@ -0,0 +1,68 @@
+from typing import Literal
+from fastapi import Depends
+from pydantic import BaseModel
+
+from ..database import db_session
+from sqlalchemy.orm import Session
+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
+
+
+class SearchResult(BaseModel):
+    type: Literal["resource", "service"]
+    data: Resource | Service
+
+
+class SearchService:
+    def __init__(self, session: Session = Depends(db_session)):
+        self._session = session
+
+    def search(self, query_str: str) -> list[SearchResult]:
+        """Searches through all tables for a string."""
+        results: list[SearchResult] = []
+        entities = (
+            ResourceEntity,
+            ServiceEntity,
+        )
+
+        for entity in entities:
+            columns = entity.__table__.columns
+
+            filters: list[BinaryExpression[bool]] = []
+            for column in columns:
+                if isinstance(column.type, String) or isinstance(column.type, 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(entity).filter(or_(*filters))
+                results.extend(
+                    [
+                        SearchResult(type=entity.__tablename__, data=result.to_model())
+                        for result in query.all()
+                    ]
+                )
+        return results