elasticsearch-dsl

import os from elasticsearch_dsl import Document, Text, Keyword, connections, Search # Configure connection to Elasticsearch # Replace with your Elasticsearch host and credentials if necessary ES_HOST = os.environ.get('ES_HOST', 'http://localhost:9200') # For cloud deployments or API key auth, uncomment and set these: # ES_CLOUD_ID = os.environ.get('ES_CLOUD_ID') # ES_API_KEY = os.environ.get('ES_API_KEY') # ES_USERNAME = os.environ.get('ES_USERNAME', 'elastic') # ES_PASSWORD = os.environ.get('ES_PASSWORD', 'changeme') connections.create_connection(hosts=[ES_HOST]) # Example for local or basic auth # For cloud/API key: connections.create_connection(cloud_id=ES_CLOUD_ID, api_key=ES_API_KEY) # Define a Document (schema for your data) class Article(Document): title = Text(fields={'keyword': Keyword()}) author = Text(fields={'keyword': Keyword()}) published_date = Keyword() word_count = Keyword() class Index: name = 'my-articles' settings = { 'number_of_shards': 1, 'number_of_replicas': 0 } # Create the index (if it doesn't exist) based on the Document definition Article.init() # Index some data article1 = Article(meta={'id': '1'}, title='Python for AI', author='John Doe', published_date='2023-01-01', word_count='2000') article1.save() article2 = Article(meta={'id': '2'}, title='Elasticsearch DSL Basics', author='Jane Smith', published_date='2023-03-15', word_count='1500') article2.save() # Refresh the index to make documents searchable immediately connections.get_connection().indices.refresh(index='my-articles') # Perform a search s = Search(index='my-articles').query("match", title="python") response = s.execute() print(f"Found {response.hits.total.value} results for 'python':") for hit in response: print(f"ID: {hit.meta.id}, Title: {hit.title}, Author: {hit.author}") # Example of a more complex search with filtering s = Search(index='my-articles') \ .query("match_all") \ .filter("range", word_count={"gte": 1500}) \ .exclude("match", author="john") response = s.execute() print(f"\nFound {response.hits.total.value} results for complex query:") for hit in response: print(f"ID: {hit.meta.id}, Title: {hit.title}, Author: {hit.author}") # Cleanup (optional) # connections.get_connection().indices.delete(index='my-articles', ignore=[400, 404])