add sqlite search, cleanup and ci

.github/workflows/lint.yaml

@@ -0,0 +1,21 @@
+name: Lint
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  lint:
+    name: Lint
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+          cache: 'pip'
+
+      - run: make install
+
+      - run: make lint

.gitignore

@@ -1,3 +1,4 @@
 venv
 /static/*.min.js
 .env
+sqlite

static/style.css

@@ -62,6 +62,7 @@ ol {
 
 form {
     margin-bottom: 1rem;
+    line-height: 3rem;
 }
 
 input[type="text"] {
@@ -83,11 +84,6 @@ input[type="text"]:focus {
     border: 1px solid #555;
 }
 
-.algolia-logo {
-    width: 20px;
-    vertical-align: baseline;
-}
-
 label {
     color: #E0E0E0;
     margin-right: 10px;
@@ -99,7 +95,7 @@ select {
     border: none;
     padding: 10px;
     border-radius: var(--border-radius);
-    min-width: min(20rem, calc(100% - var(--padding-body)));
+    /* min-width: min(20rem, calc(100% - var(--padding-body))); */
     border: 1px solid transparent;
 }
 

templates/index.jinja.html

@@ -6,21 +6,27 @@ <h1>Text search example</h1>
 
     <form
         hx-get="/html-api/items"
-        hx-params="search_query"
+        hx-params="search_query, search_index"
         hx-target="#search-results-container"
-        hx-trigger="keyup delay:800ms"
-        >
-        <input type="text" name="search_query" placeholder="Search...">
+        hx-trigger="keyup delay:800ms from:input queue:last changed, change from:select queue:last"
+    >
+        <input
+            type="text"
+            name="search_query"
+            placeholder="Search..."
+        />
 
-        <!-- TODO: add more search providers -->
-        <!-- <label for="index">Search with:</label>
-        <select id="index" name="index">
-            <option value="algolia">Algolia</option>
-            <option value="sqlite">SQLite</option>
-        </select> -->
+        <label for="search-index-select">Search with:</label>
+        <select
+            id="search-index-select"
+            name="search_index"
+            hx-trigger="change"
+        >
+            <option value="SQLITE">SQLite</option>
+            <option value="ALGOLIA">Algolia</option>
+        </select>
     </form>
 
-
     <div id="search-results-container">
         {% include "results.jinja.html" %}
     </div>

text_search_app/app.py

@@ -1,16 +1,27 @@
+from contextlib import asynccontextmanager
+
 from fastapi import FastAPI
 from fastapi.requests import Request
 from fastapi.staticfiles import StaticFiles
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from text_search_app.config import DEVELOPMENT_MODE
 from text_search_app.routers import html_api_router, index_management_router, page_router
+from text_search_app.search_indexes import sqlite_index
 from text_search_app.templates import make_template_response
 
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    sqlite_index.setup_db()
+    yield
+
+
 app = FastAPI(
-    title="Search sample",
+    title="Sample search project (FastAPI + Jinja templates + HTMLX + SQLite + Algolia)",
     debug=DEVELOPMENT_MODE,
     openapi_url="/openapi.json" if DEVELOPMENT_MODE else None,
+    lifespan=lifespan,
 )
 
 

text_search_app/config.py

@@ -10,3 +10,7 @@
 ALGOLIA_APP_ID = os.environ["ALGOLIA_APP_ID"]
 ALGOLIA_API_KEY = os.environ["ALGOLIA_API_KEY"]
 ALGOLIA_INDEX_NAME = os.environ["ALGOLIA_INDEX_NAME"]
+
+SQLITE_DATABASE_PATH = "products.sqlite"
+
+RESULTS_PAGE_SIZE = 8

text_search_app/logging.py

@@ -4,7 +4,7 @@
 
 logging.basicConfig(
     level=logging.INFO,
-    format="[%(asctime)s] [%(levelname)s] %(pathname)s:%(lineno)d | %(message)s",
+    format="[%(asctime)s] [%(levelname)s] %(message)s (%(pathname)s:%(lineno)d)",
 )
 
 logger = logging.getLogger(text_search_app.__name__)

text_search_app/models.py

@@ -1,4 +1,4 @@
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, field_validator
 
 
 class Product(BaseModel):
@@ -7,8 +7,8 @@ class Product(BaseModel):
     description: str
     price: float
     currency: str
-    terms: str
-    section: str
+    terms: str | None
+    section: str | None
 
     @field_validator("price", mode="before")
     def transform_str_to_float(
@@ -19,9 +19,3 @@ def transform_str_to_float(
             return float(value.replace(",", "."))
 
         return value
-
-
-class AlgoliaSearchResult(Product):
-    object_id: str = Field(
-        alias="objectId",
-    )

text_search_app/routers/html_api_router.py

@@ -2,8 +2,8 @@
 from fastapi.responses import HTMLResponse
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
-from text_search_app.search_indexes.algolia import get_items
 from text_search_app.config import DEVELOPMENT_MODE
+from text_search_app.search_indexes.indexes import SearchIndex, get_items
 from text_search_app.templates import make_template_response
 
 
@@ -21,7 +21,9 @@ async def prevent_html_api_direct_access(
         return
 
     # Simulate a 404 just for demonstration - I'm avoiding custom errors in this project
-    raise StarletteHTTPException(status_code=status.HTTP_404_NOT_FOUND)
+    raise StarletteHTTPException(
+        status_code=status.HTTP_404_NOT_FOUND,
+    )
 
 
 router = APIRouter(
@@ -37,14 +39,22 @@ async def prevent_html_api_direct_access(
 )
 async def get_items_html(
     request: Request,
+    search_index: SearchIndex = Query(
+        default=SearchIndex.SQLITE,
+        description="The search index to use",
+    ),
     search_query: str = Query(
         default="",
+        description="The user query to find products that better match it",
     ),
 ):
     return make_template_response(
         "results",
         request,
         {
-            "items": get_items(search_query),
+            "items": get_items(
+                search_index=search_index,
+                search_query=search_query,
+            )
         },
     )

text_search_app/routers/index_management_router.py

@@ -3,9 +3,9 @@
 from fastapi import APIRouter, File, UploadFile
 from fastapi.responses import PlainTextResponse
 
-from text_search_app.search_indexes.algolia import load_products_into_index
 from text_search_app.logging import logger
 from text_search_app.models import Product
+from text_search_app.search_indexes import indexes
 
 router = APIRouter(
     prefix="/index",
@@ -36,7 +36,7 @@ async def upload_products(
         for row in reader
     ]
 
-    load_products_into_index(products)
+    indexes.load_products_into_index(products)
 
     return PlainTextResponse(
         content=f"Successfully indexed {len(products)} products.",

text_search_app/routers/page_router.py

@@ -1,10 +1,10 @@
 from typing import Literal
 
-from fastapi import APIRouter, Query, Response
+from fastapi import APIRouter, Response
 from fastapi.requests import Request
 from fastapi.responses import HTMLResponse
 
-from text_search_app.search_indexes.algolia import get_items
+from text_search_app.search_indexes import indexes
 from text_search_app.templates import make_template_response
 
 router = APIRouter(
@@ -42,14 +42,14 @@ async def get_robots_txt():
 )
 async def get_index_html(
     request: Request,
-    search_query: str = Query(
-        default="",
-    ),
+    # TODO: later one could add a search_query query params to preserve the user state
 ):
     return make_template_response(
         "index",
         request,
         {
-            "items": get_items(search_query),
+            "items": indexes.get_items(
+                search_query=None,
+            ),
         },
     )

text_search_app/search_indexes/algolia.py → text_search_app/search_indexes/algolia_index.py

@@ -1,8 +1,13 @@
 from algoliasearch.search_client import SearchClient
 
-from text_search_app.config import ALGOLIA_API_KEY, ALGOLIA_APP_ID, ALGOLIA_INDEX_NAME
+from text_search_app.config import (
+    ALGOLIA_API_KEY,
+    ALGOLIA_APP_ID,
+    ALGOLIA_INDEX_NAME,
+    RESULTS_PAGE_SIZE,
+)
 from text_search_app.logging import logger
-from text_search_app.models import AlgoliaSearchResult, Product
+from text_search_app.models import Product
 
 algolia_client = SearchClient.create(
     ALGOLIA_APP_ID,
@@ -12,19 +17,15 @@
 algolia_index = algolia_client.init_index(ALGOLIA_INDEX_NAME)
 
 
-def get_items(search_query: str) -> list[AlgoliaSearchResult] | None:
-    """
-    Given a search query returns matching items.
-    Returns None if the index is not yet initialized.
-    """
+def get_items(search_query: str | None = None) -> list[Product] | None:
     if not algolia_index.exists():
         return None
 
     results = algolia_index.search(
         search_query,
         {
-            "page": 1,
-            "hitsPerPage": 12,
+            "page": 0,
+            "hitsPerPage": RESULTS_PAGE_SIZE,
         },
     )
 
@@ -35,17 +36,14 @@ def get_items(search_query: str) -> list[AlgoliaSearchResult] | None:
     )
 
     return [
-        AlgoliaSearchResult.model_validate(search_result)  # fmt: skip
+        Product.model_validate(search_result)  # fmt: skip
         for search_result in search_results
     ]
 
 
 def load_products_into_index(products: list[Product]):
-    """
-    Loads the provided products into the index, replacing existing ones.
-    This is a destructive action.
-    TODO: This is thread blocking since it is using Algolia's sync methods, change to async
-    """
+    # TODO: This is thread blocking since it is using Algolia's sync methods, change to async
+
     logger.info(f"Replacing all objects in Algolia {algolia_index.name!r} index")
 
     algolia_index.replace_all_objects(
@@ -59,5 +57,6 @@ def load_products_into_index(products: list[Product]):
     )
 
     logger.info(
-        f"Successfully replaced objects in {algolia_index.name!r} with {len(products)} items"
+        f"Successfully replaced objects in Algolia index {algolia_index.name!r} "
+        f"with {len(products)} items"
     )

text_search_app/search_indexes/indexes.py

@@ -0,0 +1,35 @@
+from enum import StrEnum
+
+from text_search_app.models import Product
+from text_search_app.search_indexes import algolia_index, sqlite_index
+
+
+class SearchIndex(StrEnum):
+    ALGOLIA = "ALGOLIA"
+    SQLITE = "SQLITE"
+
+
+def get_items(
+    search_index: SearchIndex = SearchIndex.SQLITE,
+    search_query: str | None = None,
+) -> list[Product] | None:
+    """
+    Given a search query returns matching items.
+    Returns None if the index is not yet initialized.
+    """
+    search_func = (
+        algolia_index.get_items  # fmt: skip
+        if search_index == SearchIndex.ALGOLIA
+        else sqlite_index.get_items
+    )
+
+    return search_func(search_query)
+
+
+def load_products_into_index(products: list[Product]):
+    """
+    Loads the provided products into the index, replacing existing ones.
+    This is a destructive action.
+    """
+    sqlite_index.load_products_into_index(products)
+    algolia_index.load_products_into_index(products)