tessl/pypi-opensearch-py
Python client for OpenSearch providing comprehensive search, indexing, and cluster management capabilities
—
Pending
dsl-queries.mddocs/
DSL and Query Building
Domain-specific language for building complex search queries, aggregations, and filters with a Pythonic API that abstracts OpenSearch's JSON-based query syntax. The DSL provides type-safe query construction and chainable operations for intuitive query building.
Capabilities
Search Builder
Central class for constructing and executing search requests with fluent interface.
class Search:
def __init__(self, using=None, index=None, doc_type=None):
"""
Initialize search object.
Parameters:
- using: OpenSearch client instance
- index (str/list): Index name(s) to search
- doc_type (str/list): Document type(s) (deprecated)
"""
def query(self, query):
"""Add query to search. Returns new Search instance."""
def filter(self, filter):
"""Add filter to search. Returns new Search instance."""
def post_filter(self, post_filter):
"""Add post filter to search. Returns new Search instance."""
def exclude(self, query):
"""Add must_not query. Returns new Search instance."""
def aggs(self, aggs):
"""Add aggregations. Returns new Search instance."""
def sort(self, *keys):
"""Add sorting. Returns new Search instance."""
def source(self, fields=None, **kwargs):
"""Configure source filtering. Returns new Search instance."""
def highlight(self, fields, **kwargs):
"""Add highlighting. Returns new Search instance."""
def suggest(self, name, text, **kwargs):
"""Add suggestion. Returns new Search instance."""
def script_fields(self, **fields):
"""Add script fields. Returns new Search instance."""
def params(self, **kwargs):
"""Set search parameters. Returns new Search instance."""
def execute(self, ignore_cache=False):
"""Execute search and return Response object."""
def scan(self):
"""Return generator for scanning all results."""
def count(self):
"""Return document count for the query."""
def delete(self):
"""Delete documents matching the query."""Query DSL
Query builders for different types of search queries.
class Q:
@classmethod
def match(cls, field=None, query=None, **kwargs):
"""
Match query for full-text search.
Parameters:
- field (str): Field to search
- query (str): Query text
- operator (str): 'and' or 'or'
- fuzziness (str/int): Fuzziness level
- minimum_should_match (str/int): Minimum should match
Returns:
Match query object
"""
@classmethod
def match_all(cls, boost=None):
"""Match all documents query."""
@classmethod
def match_none(cls):
"""Match no documents query."""
@classmethod
def match_phrase(cls, field=None, query=None, **kwargs):
"""Match phrase query for exact phrase matching."""
@classmethod
def match_phrase_prefix(cls, field=None, query=None, **kwargs):
"""Match phrase prefix query."""
@classmethod
def multi_match(cls, query=None, fields=None, **kwargs):
"""Multi-field match query."""
@classmethod
def term(cls, field=None, value=None, **kwargs):
"""Term query for exact value matching."""
@classmethod
def terms(cls, field=None, values=None, **kwargs):
"""Terms query for multiple exact values."""
@classmethod
def range(cls, field=None, **kwargs):
"""Range query for numeric/date ranges."""
@classmethod
def exists(cls, field=None):
"""Exists query to check field presence."""
@classmethod
def missing(cls, field=None):
"""Missing query to check field absence."""
@classmethod
def bool(cls, must=None, must_not=None, should=None, filter=None, **kwargs):
"""Boolean query for combining multiple queries."""
@classmethod
def nested(cls, path=None, query=None, **kwargs):
"""Nested query for nested object fields."""
@classmethod
def has_child(cls, type=None, query=None, **kwargs):
"""Has child query for parent-child relationships."""
@classmethod
def has_parent(cls, parent_type=None, query=None, **kwargs):
"""Has parent query for parent-child relationships."""
@classmethod
def prefix(cls, field=None, value=None, **kwargs):
"""Prefix query for prefix matching."""
@classmethod
def wildcard(cls, field=None, value=None, **kwargs):
"""Wildcard query with * and ? patterns."""
@classmethod
def regexp(cls, field=None, value=None, **kwargs):
"""Regular expression query."""
@classmethod
def fuzzy(cls, field=None, value=None, **kwargs):
"""Fuzzy query for approximate matching."""
@classmethod
def ids(cls, values=None, **kwargs):
"""IDs query for specific document IDs."""
@classmethod
def constant_score(cls, query=None, boost=None):
"""Constant score query."""
@classmethod
def dis_max(cls, queries=None, **kwargs):
"""Disjunction max query."""
@classmethod
def function_score(cls, query=None, functions=None, **kwargs):
"""Function score query for custom scoring."""
@classmethod
def more_like_this(cls, fields=None, like=None, **kwargs):
"""More like this query for similar documents."""
@classmethod
def simple_query_string(cls, query=None, **kwargs):
"""Simple query string query."""
@classmethod
def query_string(cls, query=None, **kwargs):
"""Full query string query with Lucene syntax."""Aggregations DSL
Builders for various types of aggregations and analytics.
class A:
@classmethod
def terms(cls, field=None, **kwargs):
"""Terms aggregation for grouping by field values."""
@classmethod
def date_histogram(cls, field=None, **kwargs):
"""Date histogram aggregation for time-based grouping."""
@classmethod
def histogram(cls, field=None, **kwargs):
"""Histogram aggregation for numeric ranges."""
@classmethod
def range(cls, field=None, **kwargs):
"""Range aggregation for custom ranges."""
@classmethod
def date_range(cls, field=None, **kwargs):
"""Date range aggregation."""
@classmethod
def nested(cls, path=None):
"""Nested aggregation for nested objects."""
@classmethod
def reverse_nested(cls, path=None):
"""Reverse nested aggregation."""
@classmethod
def filter(cls, filter=None):
"""Filter aggregation."""
@classmethod
def filters(cls, filters=None, **kwargs):
"""Filters aggregation for multiple filters."""
@classmethod
def global_agg(cls):
"""Global aggregation."""
@classmethod
def missing(cls, field=None):
"""Missing aggregation for documents without field."""
@classmethod
def sampler(cls, shard_size=None):
"""Sampler aggregation for sampling documents."""
@classmethod
def significant_terms(cls, field=None, **kwargs):
"""Significant terms aggregation."""
@classmethod
def cardinality(cls, field=None, **kwargs):
"""Cardinality metric aggregation."""
@classmethod
def avg(cls, field=None, **kwargs):
"""Average metric aggregation."""
@classmethod
def max(cls, field=None, **kwargs):
"""Max metric aggregation."""
@classmethod
def min(cls, field=None, **kwargs):
"""Min metric aggregation."""
@classmethod
def sum(cls, field=None, **kwargs):
"""Sum metric aggregation."""
@classmethod
def value_count(cls, field=None, **kwargs):
"""Value count metric aggregation."""
@classmethod
def stats(cls, field=None, **kwargs):
"""Stats metric aggregation (count, min, max, avg, sum)."""
@classmethod
def extended_stats(cls, field=None, **kwargs):
"""Extended stats metric aggregation."""
@classmethod
def percentiles(cls, field=None, **kwargs):
"""Percentiles metric aggregation."""
@classmethod
def percentile_ranks(cls, field=None, **kwargs):
"""Percentile ranks metric aggregation."""
@classmethod
def top_hits(cls, **kwargs):
"""Top hits metric aggregation for sample documents."""Multi Search
Execute multiple searches in a single request.
class MultiSearch:
def __init__(self, using=None, index=None):
"""Initialize multi-search object."""
def add(self, search):
"""Add search to multi-search. Returns self."""
def execute(self, ignore_cache=False, raise_on_error=True):
"""Execute all searches and return list of responses."""Usage Examples
Basic Query Building
from opensearchpy import Search, Q
# Create search object
s = Search(using=client, index='products')
# Add match query
s = s.query(Q('match', title='laptop'))
# Add filter
s = s.filter(Q('range', price={'gte': 100, 'lte': 1000}))
# Add sorting
s = s.sort('-price', 'title.keyword')
# Execute search
response = s.execute()
for hit in response:
print(f"{hit.title}: ${hit.price}")Complex Boolean Queries
# Build complex boolean query
complex_query = Q('bool',
must=[
Q('match', title='laptop'),
Q('range', price={'gte': 500})
],
should=[
Q('match', brand='Apple'),
Q('match', brand='Dell')
],
must_not=[
Q('term', status='discontinued')
],
filter=[
Q('term', category='electronics'),
Q('exists', field='in_stock')
],
minimum_should_match=1
)
s = Search(using=client, index='products')
s = s.query(complex_query)
response = s.execute()Aggregations
from opensearchpy import A
# Terms aggregation with sub-aggregations
s = Search(using=client, index='sales')
s.aggs.bucket('categories', A('terms', field='category.keyword', size=10)) \
.metric('avg_price', A('avg', field='price')) \
.metric('total_sales', A('sum', field='amount'))
# Date histogram aggregation
s.aggs.bucket('sales_over_time',
A('date_histogram',
field='order_date',
calendar_interval='1M',
format='yyyy-MM'
)
).metric('monthly_revenue', A('sum', field='amount'))
response = s.execute()
# Process aggregation results
for bucket in response.aggregations.categories.buckets:
print(f"Category: {bucket.key}")
print(f" Count: {bucket.doc_count}")
print(f" Avg Price: ${bucket.avg_price.value:.2f}")
print(f" Total Sales: ${bucket.total_sales.value:.2f}")Nested Queries
# Query nested objects
nested_query = Q('nested',
path='reviews',
query=Q('bool',
must=[
Q('range', **{'reviews.rating': {'gte': 4}}),
Q('match', **{'reviews.content': 'excellent'})
]
)
)
s = Search(using=client, index='products')
s = s.query(nested_query)
response = s.execute()Function Score Queries
# Custom scoring with function score
function_score_query = Q('function_score',
query=Q('match', title='laptop'),
functions=[
{
'filter': Q('term', featured=True),
'weight': 2.0
},
{
'field_value_factor': {
'field': 'popularity_score',
'factor': 1.5,
'modifier': 'log1p'
}
},
{
'gauss': {
'price': {
'origin': 500,
'scale': 200,
'decay': 0.5
}
}
}
],
score_mode='sum',
boost_mode='multiply'
)
s = Search(using=client, index='products')
s = s.query(function_score_query)
response = s.execute()Multi-Search Operations
from opensearchpy import MultiSearch
# Create multiple searches
ms = MultiSearch(using=client)
# Add different searches
ms = ms.add(Search(index='products').query(Q('match', category='laptops')))
ms = ms.add(Search(index='products').query(Q('match', category='phones')))
ms = ms.add(Search(index='orders').query(Q('range', order_date={'gte': 'now-1d'})))
# Execute all searches
responses = ms.execute()
for i, response in enumerate(responses):
print(f"Search {i+1}: {response.hits.total.value} hits")Search Highlighting
# Add highlighting to search
s = Search(using=client, index='articles')
s = s.query(Q('match', content='machine learning'))
s = s.highlight('content', fragment_size=150, number_of_fragments=3)
s = s.highlight('title', fragment_size=0, number_of_fragments=0)
response = s.execute()
for hit in response:
print(f"Title: {hit.title}")
if hasattr(hit.meta, 'highlight'):
if 'title' in hit.meta.highlight:
print(f"Highlighted title: {hit.meta.highlight.title[0]}")
if 'content' in hit.meta.highlight:
for fragment in hit.meta.highlight.content:
print(f" Fragment: {fragment}")Suggestions
# Add suggestions to search
s = Search(using=client, index='products')
s = s.suggest('title_suggestion', 'lapto', term={'field': 'title'})
s = s.suggest('completion_suggestion', 'lap', completion={'field': 'suggest'})
response = s.execute()
# Process suggestions
if hasattr(response, 'suggest'):
for suggestion in response.suggest.title_suggestion:
for option in suggestion.options:
print(f"Suggestion: {option.text} (score: {option.score})")Install with Tessl CLI
npx tessl i tessl/pypi-opensearch-py