Tutorials
Let’s have a typical search request written directly as adict:
from elasticsearch import Elasticsearchclient = Elasticsearch("https://localhost:9200")response = client.search( index="my-index", body={ "query": { "bool": { "must": [{"match": {"title": "python"}}], "must_not": [{"match": {"description": "beta"}}], "filter": [{"term": {"category": "search"}}] } }, "aggs" : { "per_tag": { "terms": {"field": "tags"}, "aggs": { "max_lines": {"max": {"field": "lines"}} } } } })for hit in response['hits']['hits']: print(hit['_score'], hit['_source']['title'])for tag in response['aggregations']['per_tag']['buckets']: print(tag['key'], tag['max_lines']['value'])The problem with this approach is that it is very verbose, prone to syntax mistakes like incorrect nesting, hard to modify (eg. adding another filter) and definitely not fun to write.
Let’s rewrite the example using the DSL module:
from elasticsearch import Elasticsearchfrom elasticsearch.dsl import Search, query, aggsclient = Elasticsearch("https://localhost:9200")s = Search(using=client, index="my-index") \ .query(query.Match("title", "python")) \ .filter(query.Term("category", "search")) \ .exclude(query.Match("description", "beta"))s.aggs.bucket('per_tag', aggs.Terms(field="tags")) \ .metric('max_lines', aggs.Max(field='lines'))response = s.execute()for hit in response: print(hit.meta.score, hit.title)for tag in response.aggregations.per_tag.buckets: print(tag.key, tag.max_lines.value)As you see, the DSL module took care of:
- creating appropriate
Queryobjects from classes - composing queries into a compound
boolquery - putting the
termquery in a filter context of theboolquery - providing a convenient access to response data
- no curly or square brackets everywhere
Let’s have a simple Python class representing an article in a blogging system:
from datetime import datetimefrom elasticsearch.dsl import Document, Date, Integer, Keyword, Text, connections# Define a default Elasticsearch clientconnections.create_connection(hosts="https://localhost:9200")class Article(Document): title: str = mapped_field(Text(analyzer='snowball', fields={'raw': Keyword()})) body: str = mapped_field(Text(analyzer='snowball')) tags: str = mapped_field(Keyword()) published_from: datetime lines: int class Index: name = 'blog' settings = { "number_of_shards": 2, } def save(self, **kwargs): self.lines = len(self.body.split()) return super(Article, self).save(** kwargs) def is_published(self): return datetime.now() > self.published_from# create the mappings in elasticsearchArticle.init()# create and save and articlearticle = Article(meta={'id': 42}, title='Hello world!', tags=['test'])article.body = ''' looong text '''article.published_from = datetime.now()article.save()article = Article.get(id=42)print(article.is_published())# Display cluster healthprint(connections.get_connection().cluster.health())In this example you can see:
- providing a default connection
- defining fields with Python type hints and additional mapping configuration when necessary
- setting index name
- defining custom methods
- overriding the built-in
.save()method to hook into the persistence life cycle - retrieving and saving the object into Elasticsearch
- accessing the underlying client for other APIs
You can see more in thepersistence chapter.
If you have yourDocuments defined you can very easily create a faceted search class to simplify searching and filtering.
from elasticsearch.dsl import FacetedSearch, TermsFacet, DateHistogramFacetclass BlogSearch(FacetedSearch): doc_types = [Article, ] # fields that should be searched fields = ['tags', 'title', 'body'] facets = { # use bucket aggregations to define facets 'tags': TermsFacet(field='tags'), 'publishing_frequency': DateHistogramFacet(field='published_from', interval='month') }# empty searchbs = BlogSearch()response = bs.execute()for hit in response: print(hit.meta.score, hit.title)for (tag, count, selected) in response.facets.tags: print(tag, ' (SELECTED):' if selected else ':', count)for (month, count, selected) in response.facets.publishing_frequency: print(month.strftime('%B %Y'), ' (SELECTED):' if selected else ':', count)You can find more details in thefaceted_search chapter.
Let’s resume the simple example of articles on a blog, and let’s assume that each article has a number of likes. For this example, imagine we want to increment the number of likes by 1 for all articles that match a certain tag and do not match a certain description. Writing this as adict, we would have the following code:
from elasticsearch import Elasticsearchclient = Elasticsearch()response = client.update_by_query( index="my-index", body={ "query": { "bool": { "must": [{"match": {"tag": "python"}}], "must_not": [{"match": {"description": "beta"}}] } }, "script"={ "source": "ctx._source.likes++", "lang": "painless" } }, )Using the DSL, we can now express this query as such:
from elasticsearch import Elasticsearchfrom elasticsearch.dsl import Search, UpdateByQueryfrom elasticsearch.dsl.query import Matchclient = Elasticsearch()ubq = UpdateByQuery(using=client, index="my-index") \ .query(Match("title", "python")) \ .exclude(Match("description", "beta")) \ .script(source="ctx._source.likes++", lang="painless")response = ubq.execute()As you can see, theUpdate By Query object provides many of the savings offered by theSearch object, and additionally allows one to update the results of the search based on a script assigned in the same manner.
You don’t have to port your entire application to get the benefits of the DSL module, you can start gradually by creating aSearch object from your existingdict, modifying it using the API and serializing it back to adict:
body = {...}# Convert to Search objects = Search.from_dict(body)# Add some filters, aggregations, queries, ...s.filter(query.Term("tags", "python"))# Convert back to dict to plug back into existing codebody = s.to_dict()- insert complicated query here