Add search route (#73)

* Add search route

* Add check for search service enabled

Author: prajwal-pai77

app/db.py

@@ -5,8 +5,16 @@
 from couchbase.exceptions import CouchbaseException
 from datetime import timedelta
 from dotenv import load_dotenv
+from couchbase.result import PingResult
+from couchbase.diagnostics import PingState, ServiceType
 import os
+import json
 from functools import cache
+from couchbase.management.search import SearchIndex
+from couchbase.exceptions import QueryIndexAlreadyExistsException
+from couchbase.options import SearchOptions
+from couchbase.search import MatchQuery, ConjunctionQuery, TermQuery
+import couchbase.search as search
 
 
 class CouchbaseClient(object):
@@ -19,6 +27,7 @@ def __init__(self, conn_str: str, username: str, password: str) -> CouchbaseClie
         self.conn_str = conn_str
         self.username = username
         self.password = password
+        self.index_name = "hotel_search"
         self.bucket_name = "travel-sample"
         self.scope_name = "inventory"
         self.connect()
@@ -57,6 +66,13 @@ def connect(self) -> None:
 
             # get a reference to our scope
             self.scope = self.bucket.scope(self.scope_name)
+            # Call the method to create the fts index if search service is enabled
+            if self.is_search_service_enabled():
+                self.create_search_index()
+            else:
+                print(
+                    "Search service is not enabled on this cluster. Skipping search index creation."
+                )
 
     def check_scope_exists(self) -> bool:
         """Check if the scope exists in the bucket"""
@@ -70,6 +86,36 @@ def check_scope_exists(self) -> bool:
                 "Error fetching scopes in cluster. \nEnsure that travel-sample bucket exists."
             )
 
+    def is_search_service_enabled(self, min_nodes: int = 1) -> bool:
+        try:
+            ping_result: PingResult = self.cluster.ping()
+            search_endpoints = ping_result.endpoints[ServiceType.Search]
+            available_search_nodes = 0
+            for endpoint in search_endpoints:
+                if endpoint.state == PingState.OK:
+                    available_search_nodes += 1
+            return available_search_nodes >= min_nodes
+        except Exception as e:
+            print(
+                f"Error checking search service status. \nEnsure that Search Service is enabled: {e}"
+            )
+            return False
+
+    def create_search_index(self) -> None:
+        """Upsert a fts index in the Couchbase cluster"""
+        try:
+            scope_index_manager = self.bucket.scope(self.scope_name).search_indexes()
+            with open(f"{self.index_name}_index.json", "r") as f:
+                index_definition = json.load(f)
+
+            # Upsert the index
+            scope_index_manager.upsert_index(SearchIndex.from_json(index_definition))
+            print(f"Index '{self.index_name}' created or updated successfully.")
+        except QueryIndexAlreadyExistsException:
+            print(f"Index with name '{self.index_name}' already exists")
+        except Exception as e:
+            print(f"Error upserting index '{self.index_name}': {e}")
+
     def close(self) -> None:
         """Close the connection to the Couchbase cluster"""
         if self.cluster:
@@ -100,6 +146,59 @@ def query(self, sql_query, *options, **kwargs):
         # kwargs are used for named parameters
         return self.scope.query(sql_query, *options, **kwargs)
 
+    def search_by_name(self, name):
+        """Perform a full-text search for hotel names using the given name"""
+        try:
+            searchQuery = search.SearchRequest.create(
+                search.MatchQuery(name, field="name")
+            )
+            searchResult = self.scope.search(
+                self.index_name, searchQuery, SearchOptions(limit=50, fields=["name"])
+            )
+            names = []
+            for row in searchResult.rows():
+                hotel = row.fields
+                names.append(hotel.get("name", ""))
+        except Exception as e:
+            print("Error while performing fts search", {e})
+        return names
+
+    def filter(self, filter: dict, limit, offset):
+        """Perform a full-text search with filters and pagination"""
+        try:
+            conjuncts = []
+
+            match_query_terms = ["description", "name", "title"]
+            conjuncts.extend(
+                [
+                    MatchQuery(filter[t], field=t)
+                    for t in match_query_terms
+                    if t in filter
+                ]
+            )
+            term_query_terms = ["city", "country", "state"]
+            conjuncts.extend(
+                [TermQuery(filter[t], field=t) for t in term_query_terms if t in filter]
+            )
+
+            if conjuncts:
+                query = ConjunctionQuery(*conjuncts)
+            else:
+                return []
+
+            options = SearchOptions(fields=["*"], limit=limit, skip=offset)
+
+            result = self.scope.search(
+                self.index_name, search.SearchRequest.create(query), options
+            )
+            hotels = []
+            for row in result.rows():
+                hotel = row.fields
+                hotels.append(hotel)
+        except Exception as e:
+            print("Error while performing fts search", {e})
+        return hotels
+
 
 @cache
 def get_db() -> CouchbaseClient:

app/main.py

@@ -1,7 +1,7 @@
 from fastapi import FastAPI
 from contextlib import asynccontextmanager
 from starlette.responses import RedirectResponse
-from app.routers import airline, airport, route
+from app.routers import airline, airport, route, hotel
 from app.db import get_db
 
 
@@ -43,6 +43,7 @@ async def lifespan(app: FastAPI):
 app.include_router(airline.router, tags=["airline"], prefix="/api/v1/airline")
 app.include_router(airport.router, tags=["airport"], prefix="/api/v1/airport")
 app.include_router(route.router, tags=["route"], prefix="/api/v1/route")
+app.include_router(hotel.router, tags=["hotel"], prefix="/api/v1/hotel")
 
 
 # Redirect to Swagger documentation on loading the API for demo purposes

app/routers/hotel.py

@@ -0,0 +1,98 @@
+from typing import List, Optional
+from typing_extensions import Annotated
+
+from app.db import get_db as CouchbaseClient
+from fastapi import APIRouter, Depends, HTTPException, Query
+from pydantic import BaseModel, Field
+
+router = APIRouter()
+
+HOTEL_COLLECTION = "hotel"
+NAME_DESCRIPTION = "Hotel Name"
+EXAMPLE_HOTEL_NAME = "Seal View"
+
+
+class HotelName(BaseModel):
+    """Model for Hotel Name"""
+
+    name: Annotated[str, Field(description="Hotel Name")]
+
+
+class Hotel(BaseModel):
+    """Model for Hotels"""
+
+    city: Optional[str] = Field(
+        None, examples=["Santa Margarita"], description="Hotel Name"
+    )
+    country: Optional[str] = Field(
+        None, examples=["United States"], description="Country Name"
+    )
+    description: Optional[str] = Field(
+        None, examples=["newly renovated"], description="Description"
+    )
+    name: Optional[str] = Field(None, examples=["KCL Campground"], description="Name")
+    state: Optional[str] = Field(None, examples=["California"], description="State")
+    title: Optional[str] = Field(
+        None, examples=["Carrizo Plain National Monument"], description="Title"
+    )
+
+
+@router.get(
+    "/autocomplete",
+    response_model=List[HotelName],
+    description="Search for hotels based on their name. \n\n This provides an example of using [Search operations](https://docs.couchbase.com/python-sdk/current/howtos/full-text-searching-with-sdk.html#search-queries) in Couchbase to search for a specific name using the fts index.\n\n Code: [`api/hotel.py`](https://github.com/couchbase-examples/python-quickstart/blob/main/src/api/hotel.py) \n Method: `get`",
+    responses={
+        200: {
+            "description": "List of Hotel Names",
+        },
+        500: {
+            "description": "Unexpected Error",
+        },
+    },
+)
+def hotel_autocomplete(
+    name: Annotated[
+        str,
+        Query(
+            description=NAME_DESCRIPTION,
+            examples=[EXAMPLE_HOTEL_NAME],
+            openapi_examples={"Seal View": {"value": "Seal View"}},
+        ),
+    ],
+    db=Depends(CouchbaseClient),
+) -> List[HotelName]:
+    """Hotel name with specified name"""
+    try:
+        result = db.search_by_name(name)
+        return [{"name": name} for name in result]
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")
+
+
+@router.post(
+    "/filter",
+    response_model=List[Hotel],
+    description="Filter hotels using various filters such as name, title, description, country, state and city. \n\n This provides an example of using [Search operations](https://docs.couchbase.com/python-sdk/current/howtos/full-text-searching-with-sdk.html#search-queries) in Couchbase to filter documents using the fts index.\n\n Code: [`api/hotel.py`](https://github.com/couchbase-examples/python-quickstart/blob/main/src/api/hotel.py) \n Method: `post`",
+    responses={
+        200: {
+            "description": "List of Hotel",
+        },
+        500: {
+            "description": "Unexpected Error",
+        },
+    },
+)
+def hotel_filter(
+    hotel: Optional[Hotel] = None,
+    limit: int = Query(10, description="Number of hotels to return (page size)"),
+    offset: int = Query(0, description="Number of hotels to skip (for pagination)"),
+    db=Depends(CouchbaseClient),
+) -> List[Hotel]:
+    """Hotel filter with various filters"""
+    try:
+        hotels = db.filter(
+            hotel.model_dump(exclude_none=True), limit=limit, offset=offset
+        )
+        return hotels
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Unexpected error: {e}")

app/tests/conftest.py

@@ -43,6 +43,11 @@ def route_collection():
     return "route"
 
 
+@pytest.fixture(scope="module")
+def hotel_api():
+    return f"{BASE_URI}/hotel"
+
+
 class Helpers:
     @staticmethod
     def delete_existing_document(couchbase_client, collection, key):

app/tests/test_hotel.py

@@ -0,0 +1,125 @@
+from fastapi.testclient import TestClient
+from app.main import app
+
+client = TestClient(app)
+
+
+class TestHotel:
+    def test_hotel_autocomplete_search(self, hotel_api):
+        """Test searching hotels by name."""
+        url = f"{hotel_api}/autocomplete"
+        search_term = "KCL"
+
+        response = client.get(url, params={"name": search_term})
+        assert response.status_code == 200
+
+        result = response.json()
+        assert len(result) > 0
+        assert all(search_term.lower() in hotel["name"].lower() for hotel in result)
+
+    def test_hotel_autocomplete_search_no_results(self, hotel_api):
+        """Test searching hotels by name with a term that should yield no results."""
+        url = f"{hotel_api}/autocomplete"
+        search_term = "XYZNonexistentHotel"
+
+        response = client.get(url, params={"name": search_term})
+        assert response.status_code == 200
+
+        result = response.json()
+        assert len(result) == 0
+
+    def test_hotel_autocomplete(self, hotel_api):
+        """Test the autocomplete endpoint with a valid name query parameter."""
+        url = f"{hotel_api}/autocomplete?name=sea"
+        response = client.get(url)
+        assert response.status_code == 200
+
+        result = response.json()
+        assert isinstance(result, list)
+        assert len(result) == 25
+
+    def test_hotel_filter_no_results(self, hotel_api):
+        """Test filtering with criteria that should yield no results."""
+        url = f"{hotel_api}/filter"
+        impossible_filter = {"city": "NonexistentCity", "country": "NonexistentCountry"}
+
+        response = client.post(url, json=impossible_filter)
+        assert response.status_code == 200
+
+        result = response.json()
+        assert len(result) == 0
+
+    def test_hotel_all_filter(self, hotel_api):
+        """Test filtering hotels with specific filters."""
+        url = f"{hotel_api}/filter"
+        hotel_filter = {
+            "title": "Carrizo Plain National Monument",
+            "name": "KCL Campground",
+            "country": "United States",
+            "city": "Santa Margarita",
+            "state": "California",
+            "description": "newly renovated",
+        }
+
+        expected_hotels = [
+            {
+                "title": "Carrizo Plain National Monument",
+                "name": "KCL Campground",
+                "country": "United States",
+                "city": "Santa Margarita",
+                "state": "California",
+                "description": "The campground has a gravel road, pit toilets, corrals and water for livestock. There are some well established shade trees and the facilities have just been renovated to include new fire rings with BBQ grates, lantern poles, and gravel roads and tent platforms.  Tenters, and small to medium sized campers will find the KCL a good fit.",
+            }
+        ]
+
+        response = client.post(url, json=hotel_filter)
+        assert response.status_code == 200
+
+        result = response.json()
+        assert result == expected_hotels
+
+    def test_hotel_country_filter(self, hotel_api):
+        """Test filtering hotels by country."""
+        url = f"{hotel_api}/filter"
+        country_filter = {"country": "United States"}
+
+        response = client.post(url, json=country_filter)
+        assert response.status_code == 200
+
+        result = response.json()
+        assert len(result) > 0
+        assert all(hotel["country"] == "United States" for hotel in result)
+
+    def test_hotel_with_single_filter(self, hotel_api):
+        """Test filtering hotels with a single description filter."""
+        url = f"{hotel_api}/filter"
+        hotel_filter = {"description": "newly renovated"}
+
+        response = client.post(url, json=hotel_filter)
+        assert response.status_code == 200
+
+        result = response.json()
+        assert isinstance(result, list)
+        assert len(result) > 2
+
+    def test_hotel_single_filter_with_pagination(self, hotel_api):
+        """Test filtering hotels with description, offset, and limit filters."""
+        page_size = 3
+        iterations = 3
+        hotel_filter = {"description": "newly renovated"}
+        all_hotels = set()
+
+        for i in range(iterations):
+            hotel_filter["offset"] = page_size * i
+            url = f"{hotel_api}/filter?limit={page_size}&offset={page_size*i}"
+            response = client.post(url, json=hotel_filter)
+            assert response.status_code == 200
+
+            result = response.json()
+            assert isinstance(result, list)
+            assert len(result) <= page_size
+
+            for hotel in result:
+                all_hotels.add(hotel["name"])
+
+        assert len(all_hotels) >= page_size * iterations