Feat/metadata search with postgres fts

core/migrations/0017_schema_search_vector.py

@@ -0,0 +1,26 @@
+# Generated by Django 5.2.5 on 2026-07-17 21:19
+
+import django.contrib.postgres.indexes
+import django.contrib.postgres.search
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('core', '0016_schemaref_id_value'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='schema',
+            name='search_vector',
+            field=models.GeneratedField(db_persist=True, expression=django.contrib.postgres.search.CombinedSearchVector(django.contrib.postgres.search.SearchVector('name', config='english', weight='A'), '||', django.contrib.postgres.search.SearchVector('description', config='english', weight='B'), django.contrib.postgres.search.SearchConfig('english')), output_field=django.contrib.postgres.search.SearchVectorField()),
+        ),
+        migrations.AddIndex(
+            model_name='schema',
+            index=django.contrib.postgres.indexes.GinIndex(fields=['search_vector'], name='schema_search_vec_gin'),
+        ),
+    ]

core/models.py

@@ -7,6 +7,13 @@
 from django.contrib.contenttypes.fields import GenericForeignKey, GenericRelation
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.auth.hashers import make_password, check_password
+from django.contrib.postgres.search import (
+    SearchVector,
+    SearchQuery,
+    SearchRank,
+    SearchVectorField,
+)
+from django.contrib.postgres.indexes import GinIndex
 from django.core.cache import cache
 from django.core.exceptions import ValidationError
 from django.urls import reverse
@@ -87,7 +94,30 @@ class Meta:
         indexes = [models.Index(fields=["content_type", "object_id"])]
 
 
-class PublicSchemaManager(models.Manager):
+class SchemaQuerySet(models.QuerySet):
+    def search(self, query_text):
+        """
+        Rank schemas by full-text relevance against `query_text`.
+        `name` weighted above description.
+
+        When `query_text` is blank the queryset is returned unchanged,
+        so the existing ordering (alphabetical on index) is preserved for plain browsing.
+        """
+        query_text = (query_text or "").strip()
+        if not query_text:
+            return self
+        search_query = SearchQuery(
+            query_text, config="english", search_type="websearch"
+        )
+        return (
+            self
+            .filter(search_vector=search_query)
+            .annotate(rank=SearchRank(models.F("search_vector"), search_query))
+            .order_by("-rank", "name")
+        )
+
+
+class PublicSchemaManager(models.Manager.from_queryset(SchemaQuerySet)):
     def get_queryset(self):
         return (
             super()
@@ -117,9 +147,20 @@ class Schema(BaseModel):
     published_at = models.DateTimeField(blank=True, null=True)
     permanent_urls = GenericRelation(PermanentURL, related_query_name="schema")
     description = models.CharField(blank=True, null=True, max_length=350)
+    search_vector = models.GeneratedField(
+        expression=(
+            SearchVector("name", weight="A", config="english")
+            + SearchVector("description", weight="B", config="english")
+        ),
+        output_field=SearchVectorField(),
+        db_persist=True,
+    )
 
     class Meta:
-        indexes = [models.Index(fields=["published_at"])]
+        indexes = [
+            models.Index(fields=["published_at"]),
+            GinIndex(fields=["search_vector"], name="schema_search_vec_gin"),
+        ]
 
     def __str__(self):
         return self.name

core/views.py

@@ -143,23 +143,21 @@ def index(request):
         else defined_schemas
     )
 
-    filtered_by_name = (
-        filtered_by_documentation_type.filter(name__icontains=search_query)
-        if search_query
-        else filtered_by_documentation_type
-    )
+    # Full-text ranked search. When the box is empty, .search() returns the
+    # queryset unchanged, preserving the alphabetical browse ordering above.
+    searched_schemas = filtered_by_documentation_type.search(search_query)
 
     filtered_by_specification_file_type = (
         [
             schema
-            for schema in filtered_by_name
+            for schema in searched_schemas
             if any(
                 schema_ref.language == specification_file_type
                 for schema_ref in schema.schemaref_set.all()
             )
         ]
         if specification_file_type
-        else filtered_by_name
+        else searched_schemas
     )
 
     return render(

schemaindex/settings/base.py

@@ -43,6 +43,7 @@
     "django.contrib.admin",
     "django.contrib.auth",
     "django.contrib.contenttypes",
+    "django.contrib.postgres",
     "django.contrib.sessions",
     "django.contrib.messages",
     "django.contrib.staticfiles",

tests/test_search.py

@@ -0,0 +1,75 @@
+import pytest
+
+from tests.factories import SchemaFactory
+from core.models import Schema
+
+
+@pytest.mark.django_db
+def test_stemming_makes_singular_and_plural_equivalent():
+    # "Schema" and "Schemas" must return the same result.
+    # decoy proves the @@ filter actually excludes non-matches (not just that matches are found).
+    schema = SchemaFactory(name="Invoice Schema")
+    decoy = SchemaFactory(name="Weather Report")
+    for term in ("Schemas", "Schema"):
+        results = Schema.public_objects.search(term)
+        assert schema in results
+        assert decoy not in results
+
+
+@pytest.mark.django_db
+def test_search_is_case_insensitive():
+    schema = SchemaFactory(name="Invoice Schema")
+    assert schema in Schema.public_objects.search("INVOICE")
+
+
+@pytest.mark.django_db
+def test_search_matches_description_field():
+    schema = SchemaFactory(name="ACME Format", description="A standard for invoices")
+    assert schema in Schema.public_objects.search("invoice")
+
+
+@pytest.mark.django_db
+def test_name_hit_outranks_description_only_hit():
+    # Assert on the rank *values*, not list order.
+    # Order under equal ranks is decided by the name tie-break,
+    # which would hide a weight bug for some names.
+    SchemaFactory(name="Invoice Format", description=None)
+    SchemaFactory(name="ACME Format", description="Used for invoice data")
+    ranks = {s.name: s.rank for s in Schema.public_objects.search("invoice")}
+    assert ranks["Invoice Format"] > ranks["ACME Format"]
+
+
+@pytest.mark.django_db
+def test_blank_query_returns_all_and_preserves_order():
+    SchemaFactory(name="One")
+    SchemaFactory(name="Two")
+    # A blank / None query must neither filter nor reorder: .search() returns the
+    # queryset unchanged, so the caller's ordering survives and no relevance
+    # ranking is applied. (Whitespace-only counts as blank via .strip().)
+    for blank in ("", "   ", None):
+        results = Schema.public_objects.order_by("name").search(blank)
+        assert [schema.name for schema in results] == ["One", "Two"]
+
+
+@pytest.mark.django_db
+def test_search_excludes_unpublished_schemas():
+    SchemaFactory(name="Public Invoice")
+    SchemaFactory(name="Private Invoice", published_at=None)
+    assert Schema.public_objects.search("invoice").count() == 1
+
+
+@pytest.mark.django_db
+def test_malformed_query_does_not_raise():
+    SchemaFactory(name="Invoice")
+    # websearch_to_tsquery tolerates junk such as unbalanced quote plus dangling operator.
+    # The guarantee is that evaluating the queryset does not raise.
+    results = Schema.public_objects.search('"invoice OR -')
+    assert results.count() >= 0
+
+
+@pytest.mark.django_db
+def test_generated_vector_is_populated_on_insert():
+    # Postgres computes the stored generated column on write
+    schema = SchemaFactory(name="Invoice Schema")
+    schema.refresh_from_db()
+    assert schema.search_vector is not None

tests/test_views.py

@@ -431,3 +431,21 @@ def test_schema_export_sends_manifest():
     assert response.status_code == 200
     manifest = json.loads(response.content)
     assert_schema_matches_manifest(schema, manifest)
+
+
+@pytest.mark.django_db
+def test_homepage_search_is_stemmed():
+    schema = SchemaFactory(name="Invoice Schema")
+    SchemaRefFactory(schema=schema)
+    response = Client().get("/?search_query=Schemas")
+    assert schema.name in str(response.content)
+
+
+@pytest.mark.django_db
+def test_homepage_no_query_lists_alphabetically():
+    last_schema = SchemaFactory(name="ZZZ Last Schema")
+    SchemaRefFactory(schema=last_schema)
+    first_schema = SchemaFactory(name="AAA First Schema")
+    SchemaRefFactory(schema=first_schema)
+    names = [s.name for s in Client().get("/").context["schemas"]]
+    assert names == ["AAA First Schema", "ZZZ Last Schema"]