tessl/pypi-couchbase

Python Client for Couchbase providing comprehensive database operations including key-value, N1QL queries, search, analytics, and cluster management

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Search Operations

Name: tessl/pypi-couchbase
Author: tessl

Advanced full-text search capabilities with support for complex queries, faceting, sorting, and highlighting. Enables powerful text search across document collections using Couchbase's integrated search service.

Capabilities

Search Query Execution

Execute full-text search queries with various options and result formatting.

class Cluster:
    def search_query(self, index: str, query: SearchQuery, options: SearchOptions = None) -> SearchResult:
        """
        Execute full-text search query.

        Args:
            index (str): Search index name
            query (SearchQuery): Search query object
            options (SearchOptions, optional): Search execution options

        Returns:
            SearchResult: Search results iterator

        Raises:
            SearchException: If search execution fails
            IndexNotFoundException: If search index doesn't exist
        """

class SearchOptions:
    def __init__(self, timeout: timedelta = None,
                 skip: int = None,
                 limit: int = None,
                 explain: bool = False,
                 highlight: HighlightStyle = None,
                 fields: List[str] = None,
                 facets: Dict[str, Facet] = None,
                 sort: List[Sort] = None,
                 scan_consistency: SearchScanConsistency = None,
                 consistent_with: MutationState = None,
                 raw: Dict[str, Any] = None):
        """
        Search query execution options.

        Args:
            timeout (timedelta, optional): Search timeout
            skip (int, optional): Number of results to skip
            limit (int, optional): Maximum results to return
            explain (bool): Include score explanation
            highlight (HighlightStyle, optional): Result highlighting
            fields (List[str], optional): Fields to return
            facets (Dict[str, Facet], optional): Facet definitions
            sort (List[Sort], optional): Result sorting
            scan_consistency (SearchScanConsistency, optional): Consistency level
            consistent_with (MutationState, optional): Consistency token
            raw (Dict[str, Any], optional): Raw search options
        """

Search Query Types

Various search query types for different search patterns.

class SearchQuery:
    """Base class for all search queries."""

    @staticmethod
    def match(match: str) -> MatchQuery:
        """
        Create match query for text search.

        Args:
            match (str): Text to match

        Returns:
            MatchQuery: Match query instance
        """

    @staticmethod
    def term(term: str) -> TermQuery:
        """
        Create term query for exact matches.

        Args:
            term (str): Exact term to match

        Returns:
            TermQuery: Term query instance
        """

    @staticmethod
    def phrase(terms: List[str]) -> PhraseQuery:
        """
        Create phrase query for exact phrase matching.

        Args:
            terms (List[str]): Terms in phrase

        Returns:
            PhraseQuery: Phrase query instance
        """

    @staticmethod
    def prefix(prefix: str) -> PrefixQuery:
        """
        Create prefix query.

        Args:
            prefix (str): Prefix to match

        Returns:
            PrefixQuery: Prefix query instance
        """

    @staticmethod
    def wildcard(wildcard: str) -> WildcardQuery:
        """
        Create wildcard query with * and ? patterns.

        Args:
            wildcard (str): Wildcard pattern

        Returns:
            WildcardQuery: Wildcard query instance
        """

    @staticmethod
    def regex(regexp: str) -> RegexQuery:
        """
        Create regular expression query.

        Args:
            regexp (str): Regular expression pattern

        Returns:
            RegexQuery: Regex query instance
        """

    @staticmethod
    def numeric_range(min_val: float = None, max_val: float = None) -> NumericRangeQuery:
        """
        Create numeric range query.

        Args:
            min_val (float, optional): Minimum value
            max_val (float, optional): Maximum value

        Returns:
            NumericRangeQuery: Numeric range query instance
        """

    @staticmethod
    def date_range(start: datetime = None, end: datetime = None) -> DateRangeQuery:
        """
        Create date range query.

        Args:
            start (datetime, optional): Start date
            end (datetime, optional): End date

        Returns:
            DateRangeQuery: Date range query instance
        """

    @staticmethod
    def geo_distance(location: GeoPoint, distance: str) -> GeoDistanceQuery:
        """
        Create geographic distance query.

        Args:
            location (GeoPoint): Center point
            distance (str): Distance (e.g., "10km", "5mi")

        Returns:
            GeoDistanceQuery: Geographic distance query instance
        """

    @staticmethod
    def geo_bounding_box(top_left: GeoPoint, bottom_right: GeoPoint) -> GeoBoundingBoxQuery:
        """
        Create geographic bounding box query.

        Args:
            top_left (GeoPoint): Top-left corner
            bottom_right (GeoPoint): Bottom-right corner

        Returns:
            GeoBoundingBoxQuery: Geographic bounding box query instance
        """

    @staticmethod
    def boolean() -> BooleanQuery:
        """
        Create boolean combination query.

        Returns:
            BooleanQuery: Boolean query instance
        """

    @staticmethod
    def match_all() -> MatchAllQuery:
        """
        Create match-all query.

        Returns:
            MatchAllQuery: Match all query instance
        """

    @staticmethod
    def match_none() -> MatchNoneQuery:
        """
        Create match-none query.

        Returns:
            MatchNoneQuery: Match none query instance
        """

Query Configuration

Configure individual query types with specific options.

class MatchQuery(SearchQuery):
    def field(self, field: str) -> MatchQuery:
        """Specify field to search."""

    def analyzer(self, analyzer: str) -> MatchQuery:
        """Set text analyzer."""

    def boost(self, boost: float) -> MatchQuery:
        """Set query boost factor."""

    def prefix_length(self, prefix_length: int) -> MatchQuery:
        """Set prefix length for fuzzy matching."""

    def fuzziness(self, fuzziness: int) -> MatchQuery:
        """Set fuzziness level."""

    def operator(self, operator: MatchOperator) -> MatchQuery:
        """Set match operator (AND/OR)."""

class TermQuery(SearchQuery):
    def field(self, field: str) -> TermQuery:
        """Specify field to search."""

    def boost(self, boost: float) -> TermQuery:
        """Set query boost factor."""

class NumericRangeQuery(SearchQuery):
    def field(self, field: str) -> NumericRangeQuery:
        """Specify numeric field."""

    def min(self, min_val: float, inclusive: bool = True) -> NumericRangeQuery:
        """Set minimum value."""

    def max(self, max_val: float, inclusive: bool = True) -> NumericRangeQuery:
        """Set maximum value."""

    def boost(self, boost: float) -> NumericRangeQuery:
        """Set query boost factor."""

class BooleanQuery(SearchQuery):
    def must(self, query: SearchQuery) -> BooleanQuery:
        """Add must clause (AND)."""

    def should(self, query: SearchQuery) -> BooleanQuery:
        """Add should clause (OR)."""

    def must_not(self, query: SearchQuery) -> BooleanQuery:
        """Add must not clause (NOT)."""

    def boost(self, boost: float) -> BooleanQuery:
        """Set query boost factor."""

Faceting

Aggregate search results by categories for navigation and filtering.

class Facet:
    """Base class for all facets."""

    @staticmethod
    def term(field: str, size: int = 10) -> TermFacet:
        """
        Create term facet for categorical aggregation.

        Args:
            field (str): Field to facet on
            size (int): Maximum number of facet terms

        Returns:
            TermFacet: Term facet instance
        """

    @staticmethod
    def date_range(field: str, *ranges: DateRange) -> DateRangeFacet:
        """
        Create date range facet.

        Args:
            field (str): Date field to facet on
            *ranges: Date range definitions

        Returns:
            DateRangeFacet: Date range facet instance
        """

    @staticmethod
    def numeric_range(field: str, *ranges: NumericRange) -> NumericRangeFacet:
        """
        Create numeric range facet.

        Args:
            field (str): Numeric field to facet on
            *ranges: Numeric range definitions

        Returns:
            NumericRangeFacet: Numeric range facet instance
        """

class DateRange:
    def __init__(self, name: str, start: datetime = None, end: datetime = None):
        """Date range for faceting."""

class NumericRange:
    def __init__(self, name: str, min_val: float = None, max_val: float = None):
        """Numeric range for faceting."""

Result Sorting

Sort search results by various criteria.

class Sort:
    """Base class for all sort options."""

    @staticmethod
    def score() -> SortScore:
        """Sort by relevance score (default)."""

    @staticmethod
    def id() -> SortId:
        """Sort by document ID."""

    @staticmethod
    def field(field: str) -> SortField:
        """Sort by field value."""

    @staticmethod
    def geo_distance(location: GeoPoint, field: str) -> SortGeoDistance:
        """Sort by geographic distance."""

class SortField(Sort):
    def desc(self) -> SortField:
        """Sort in descending order."""

    def type(self, type: str) -> SortField:
        """Set field type (auto, string, number, date)."""

    def mode(self, mode: str) -> SortField:
        """Set sort mode (min, max, default)."""

    def missing(self, missing: str) -> SortField:
        """Set behavior for missing values (first, last)."""

Search Results

class SearchResult:
    def __iter__(self) -> Iterator[SearchRow]:
        """Iterate over search result rows."""

    def metadata(self) -> SearchMetaData:
        """Get search execution metadata."""

    def rows(self) -> List[SearchRow]:
        """Get all result rows as list."""

    def facets(self) -> Dict[str, SearchFacetResult]:
        """Get facet results."""

class SearchRow:
    @property
    def index(self) -> str:
        """Search index name."""

    @property
    def id(self) -> str:
        """Document ID."""

    @property
    def score(self) -> float:
        """Relevance score."""

    @property
    def explanation(self) -> dict:
        """Score explanation (if requested)."""

    @property
    def locations(self) -> dict:
        """Match locations."""

    @property
    def fragments(self) -> dict:
        """Highlighted fragments."""

    @property
    def fields(self) -> dict:
        """Retrieved field values."""

class SearchMetaData:
    @property
    def metrics(self) -> SearchMetrics:
        """Search execution metrics."""

    @property
    def errors(self) -> Dict[str, str]:
        """Search execution errors."""

class SearchMetrics:
    @property
    def took(self) -> timedelta:
        """Search execution time."""

    @property
    def total_hits(self) -> int:
        """Total number of matching documents."""

    @property
    def max_score(self) -> float:
        """Maximum relevance score."""

    @property
    def success_partition_count(self) -> int:
        """Number of successful partitions."""

    @property
    def error_partition_count(self) -> int:
        """Number of failed partitions."""

Usage Examples

Basic Search Operations

from couchbase.search import SearchQuery, SearchOptions

# Simple text search
query = SearchQuery.match("hotel california")
result = cluster.search_query("travel-search", query)

for row in result:
    print(f"Document {row.id}: Score {row.score}")
    if 'name' in row.fields:
        print(f"  Name: {row.fields['name']}")

# Get search metadata
metadata = result.metadata()
print(f"Search took: {metadata.metrics.took}")
print(f"Total hits: {metadata.metrics.total_hits}")

Complex Query Construction

# Boolean query with multiple conditions
bool_query = SearchQuery.boolean()
bool_query.must(SearchQuery.match("luxury").field("description"))
bool_query.should(SearchQuery.term("hotel").field("type"))
bool_query.must_not(SearchQuery.term("closed").field("status"))

options = SearchOptions(limit=20, skip=0)
result = cluster.search_query("travel-search", bool_query, options)

Range Queries

# Numeric range query
price_query = SearchQuery.numeric_range(min_val=100, max_val=500).field("price")

# Date range query
from datetime import datetime
start_date = datetime(2023, 1, 1)
end_date = datetime(2023, 12, 31)
date_query = SearchQuery.date_range(start=start_date, end=end_date).field("created")

# Combine with boolean query
combined = SearchQuery.boolean()
combined.must(SearchQuery.match("luxury hotel"))
combined.must(price_query)
combined.must(date_query)

result = cluster.search_query("travel-search", combined)

Geographic Search

from couchbase.search import GeoPoint

# Geographic distance search
san_francisco = GeoPoint(37.7749, -122.4194)
geo_query = SearchQuery.geo_distance(san_francisco, "10km").field("location")

result = cluster.search_query("travel-search", geo_query)

for row in result:
    print(f"Hotel {row.id} within 10km of SF")

Faceted Search

from couchbase.search import Facet

# Search with facets
facets = {
    "type": Facet.term("type", size=10),
    "price_ranges": Facet.numeric_range("price",
        NumericRange("budget", max_val=100),
        NumericRange("mid_range", min_val=100, max_val=300),
        NumericRange("luxury", min_val=300)
    ),
    "ratings": Facet.numeric_range("rating",
        NumericRange("poor", max_val=2),
        NumericRange("good", min_val=2, max_val=4),
        NumericRange("excellent", min_val=4)
    )
}

options = SearchOptions(facets=facets, limit=50)
result = cluster.search_query("travel-search", SearchQuery.match_all(), options)

# Process facet results
facet_results = result.facets()
for facet_name, facet_result in facet_results.items():
    print(f"Facet: {facet_name}")
    for term in facet_result.terms:
        print(f"  {term.term}: {term.count}")

Sorting and Field Selection

from couchbase.search import Sort

# Sort by multiple criteria
sort_options = [
    Sort.field("rating").desc(),
    Sort.field("price"),
    Sort.score()
]

options = SearchOptions(
    sort=sort_options,
    fields=["name", "description", "price", "rating"],  # Only return these fields
    limit=20
)

result = cluster.search_query("travel-search", SearchQuery.match("hotel"), options)

for row in result:
    print(f"{row.fields['name']}: ${row.fields['price']}, Rating: {row.fields['rating']}")

Highlighting

from couchbase.search import HighlightStyle

# Search with result highlighting
options = SearchOptions(
    highlight=HighlightStyle.HTML,
    fields=["name", "description"]
)

query = SearchQuery.match("luxury beachfront resort")
result = cluster.search_query("travel-search", query, options)

for row in result:
    print(f"Document: {row.id}")
    if row.fragments:
        for field, fragments in row.fragments.items():
            print(f"  {field}: {' ... '.join(fragments)}")

Search with Consistency

from couchbase.mutation_state import MutationState

# Perform document update
doc = {"name": "New Luxury Hotel", "type": "hotel", "rating": 5}
mutation_result = collection.upsert("hotel::new", doc)

# Search with consistency
mutation_state = MutationState(mutation_result.mutation_token)
options = SearchOptions(consistent_with=mutation_state)

result = cluster.search_query("travel-search", 
                             SearchQuery.match("New Luxury Hotel"), 
                             options)

Error Handling

from couchbase.exceptions import SearchException, IndexNotFoundException

try:
    result = cluster.search_query("nonexistent-index", SearchQuery.match_all())
    for row in result:
        print(row.id)
except IndexNotFoundException:
    print("Search index not found")
except SearchException as e:
    print(f"Search failed: {e}")
    if hasattr(e, 'context'):
        print(f"Error details: {e.context}")

Install with Tessl CLI