search

name	search
description	Search implementation and optimization
license	MIT
compatibility	opencode
metadata	{"audience":"developers","category":"database"}

What I do

• Implement search functionality
• Use Elasticsearch effectively
• Design search indexes
• Handle search queries
• Optimize search performance
• Implement fuzzy matching
• Handle faceted search
• Implement search suggestions

When to use me

When implementing search functionality or optimizing search.

Elasticsearch Client

from elasticsearch import Elasticsearch
from typing import List, Dict, Any, Optional
from dataclasses import dataclass


@dataclass
class SearchQuery:
    """Search query parameters."""
    query: str
    filters: Dict[str, Any] = None
    sort: List[Dict] = None
    page: int = 1
    page_size: int = 10
    highlight: bool = True
    aggregations: List[str] = None


class ElasticsearchClient:
    """Elasticsearch client wrapper."""
    
    def __init__(self, hosts: List[str]) -> None:
        self.client = Elasticsearch(hosts=hosts)
    
    def search(
        self,
        index: str,
        query: SearchQuery
    ) -> Dict[str, Any]:
        """Execute search query."""
        es_query = self._build_query(query)
        
        body = {"query": es_query}
        
        if query.sort:
            body["sort"] = query.sort
        
        if query.highlight:
            body["highlight"] = {
                "fields": {
                    "content": {},
                    "title": {},
                }
            }
        
        if query.aggregations:
            body["aggs"] = self._build_aggregations(query.aggregations)
        
        # Pagination
        from_index = (query.page - 1) * query.page_size
        body["from"] = from_index
        body["size"] = query.page_size
        
        return self.client.search(index=index, body=body)
    
    def _build_query(self, query: SearchQuery) -> Dict[str, Any]:
        """Build Elasticsearch query."""
        must = []
        filter_clauses = []
        
        # Full-text search
        if query.query:
            must.append({
                "multi_match": {
                    "query": query.query,
                    "fields": ["title^3", "content^2", "tags", "description"],
                    "type": "best_fields",
                    "fuzziness": "AUTO",
                }
            })
        
        # Filters
        if query.filters:
            for field, value in query.filters.items():
                if isinstance(value, list):
                    filter_clauses.append({
                        "terms": {field: value}
                    })
                else:
                    filter_clauses.append({
                        "term": {field: value}
                    })
        
        # Combine
        es_query = {
            "bool": {
                "must": must if must else [{"match_all": {}}],
                "filter": filter_clauses,
            }
        }
        
        return es_query
    
    def _build_aggregations(self, fields: List[str]) -> Dict[str, Any]:
        """Build aggregations for faceted search."""
        aggs = {}
        
        for field in fields:
            aggs[field] = {
                "terms": {
                    "field": field,
                    "size": 20,
                }
            }
        
        return aggs


# Usage
es_client = ElasticsearchClient(["localhost:9200"])

results = es_client.search(
    index="products",
    query=SearchQuery(
        query="wireless headphones",
        filters={"category": ["electronics"], "brand": ["sony"]},
        sort=[{"price": {"order": "asc"}}],
        page=1,
        page_size=20,
        aggregations=["category", "brand", "price_range"],
    )
)

Full-Text Search

-- PostgreSQL full-text search

-- Create tsvector column
ALTER TABLE products
ADD COLUMN search_vector tsvector;

-- Update search vector
UPDATE products
SET search_vector = 
    setweight(to_tsvector('english', COALESCE(name, '')), 'A') ||
    setweight(to_tsvector('english', COALESCE(description, '')), 'B') ||
    setweight(to_tsvector('english', COALESCE(tags, '')), 'C');

-- Create index
CREATE INDEX idx_products_search
ON products USING GIN(search_vector);

-- Search query
SELECT id, name, ts_rank(search_vector, query) AS rank
FROM products,
     to_tsquery('english', 'wireless & headphones') query
WHERE search_vector @@ query
ORDER BY rank DESC
LIMIT 20;


-- With ranking and highlighting
SELECT 
    id,
    name,
    ts_rank_cd(search_vector, query) AS rank,
    ts_headline('english', description, query) AS highlighted_description
FROM products,
     to_tsquery('english', 'wireless <-> headphones') query
WHERE search_vector @@ query
  rank ;

Search Suggestions

from typing import List, Tuple


class SearchSuggestions:
    """Implement search autocomplete/suggestions."""
    
    def __init__(self, es_client: ElasticsearchClient) -> None:
        self.client = es_client
    
    def get_suggestions(
        self,
        query: str,
        size: int = 10,
        field: str = "name.suggest"
    ) -> List[str]:
        """Get search suggestions for autocomplete."""
        if len(query) < 2:
            return []
        
        es_query = {
            "suggest": {
                "product-suggest": {
                    "prefix": query,
                    "completion": {
                        "field": field,
                        "size": size,
                        "skip_duplicates": True,
                        "fuzzy": {
                            "fuzziness": "AUTO",
                        }
                    }
                }
            }
        }
        
        result = self.client.client.suggest(
            index="products",
            body=es_query
        )
        
        suggestions = []
        for suggestion in result['suggest'][][][]:
            suggestions.append(suggestion[])
        
         suggestions
    
     () -> [[, ]]:
        
        
         [
            (, ),
            (, ),
            (, ),
            (, ),
            (, ),
        ]
    
     () -> []:
        
        
         []

Faceted Search

@dataclass
class FacetedSearchResult:
    """Faceted search result with aggregations."""
    total_hits: int
    page: int
    page_size: int
    results: List[Dict[str, Any]]
    facets: Dict[str, List[Dict[str, Any]]]
    query: str


class FacetedSearch:
    """Implement faceted search."""
    
    def __init__(self, es_client: ElasticsearchClient) -> None:
        self.client = es_client
    
    def search(
        self,
        index: str,
        query: str,
        facet_fields: List[str],
        page: int = 1,
        page_size: int = 20,
        filters: Dict[str, Any] = None,
    ) -> FacetedSearchResult:
        """Execute faceted search."""
        
        # Build query
        es_query = self._build_faceted_query(query, filters)
        
        # Build aggregations
        aggs = {}
        for field in facet_fields:
            aggs[field] = {
                : {
                    : field,
                    : ,
                }
            }
        
        
        body = {
            : es_query,
            : aggs,
            : (page - ) * page_size,
            : page_size,
        }
        
        result = .client.client.search(index=index, body=body)
        
        
        hits = result[][][]
        results = [hit[]  hit  result[][]]
        
        
        facets = {}
         field  facet_fields:
            facets[field] = [
                {
                    : bucket[],
                    : bucket[],
                }
                 bucket  result[][field][]
            ]
        
         FacetedSearchResult(
            total_hits=hits,
            page=page,
            page_size=page_size,
            results=results,
            facets=facets,
            query=query,
        )
    
     () -> [, ]:
        
        must = []
        filter_clauses = []
        
         query:
            must.append({
                : {
                    : query,
                    : [, ],
                }
            })
        
         filters:
             field, value  filters.items():
                 value:  
                    filter_clauses.append({
                        : {field: value}
                    })
        
         {
            : {
                : must  must  [{: {}}],
                : filter_clauses,
            }
        }

Search Ranking

class SearchRanker:
    """Custom search ranking factors."""
    
    # Weights for different factors
    WEIGHTS = {
        "text_relevance": 1.0,
        "popularity": 0.3,
        "recency": 0.2,
        "rating": 0.1,
        "price": -0.1,
    }
    
    def boost_results(
        self,
        results: List[Dict[str, Any]],
        query: str
    ) -> List[Dict[str, Any]]:
        """Apply ranking boosts to results."""
        
        for result in results:
            boost_score = 0.0
            
            # Text relevance (from Elasticsearch)
            boost_score += result.get('_score', 0) * self.WEIGHTS['text_relevance']
            
            # Popularity boost
            popularity = result.get('view_count', 0) + result.get('sales_count', 0) * 10
            boost_score += self._normalize(popularity, 10000) * self.WEIGHTS['popularity']
            
            # Recency boost
            days_since = (datetime.utcnow() - result.get(, datetime.)).days
            recency_score = (,  - days_since / )  
            boost_score += recency_score * .WEIGHTS[]
            
            
            avg_rating = result.get(, )
            boost_score += (avg_rating / ) * .WEIGHTS[]
            
            
               result:
                price_factor =  / ( + result[] / )
                boost_score += price_factor * .WEIGHTS[]
            
            result[] = boost_score
        
        
        results.sort(key= x: x.get(, ), reverse=)
        
         results
    
     () -> :
        
         (, value / max_value)

Best Practices

1. Design indexes for queries
   - Consider access patterns
   - Use appropriate analyzers

2. Optimize for search speed
   - Use covering indexes
   - Limit returned fields

3. Handle synonyms
   - Map synonyms to common terms
   - Build synonym dictionary

4. Implement pagination
   - Use cursor-based for large datasets
   - Limit page sizes

5. Handle edge cases
   - No results query
   - Misspellings (fuzzy matching)
   - Special characters

6. Secure search
   - Filter by user permissions
   - No information leakage

7. Monitor performance
   - Track query latency
   - Monitor index size

8. Test with real data
   - Test with common queries
   - Test edge cases
   - Test performance under load