Autocomplete

elasticsearch_repo.py

def create_index(es_object, index_name="recipes"):
    created = False
    # index settings
    settings = {
        "settings": {"number_of_shards": 1, "number_of_replicas": 0},
        "mappings": {
            "members": {
                "dynamic": "strict",
                "properties": {
                    "title": {"type": "text"},
                    "slug": {"type": "text"},
                    "description": {"type": "text"},
                },
            }
        },
    }
    try:
        if not current_app.elasticsearch.indices.exists(index_name):
            # Ignore 400 means to ignore "Index Already Exist" error.
            current_app.elasticsearch.indices.create(
                index=index_name, ignore=400, body=settings
            )
            print("Created Index")
        created = True
    except Exception as ex:
        print(str(ex))
    finally:
        return created


def add_to_index(index, payload):
    if not current_app.elasticsearch:
        return

    current_app.elasticsearch.index(
        index=index, doc_type=index, id=f"{payload['id']}", body=payload
    )


def update_index(index, payload):
    if not current_app.elasticsearch:
        return

    current_app.elasticsearch.update(
        index=index, doc_type=index, id=f"{payload.id}", body={"doc": payload}
    )


def remove_from_index(index, object_id):
    if not current_app.elasticsearch:
        return

    current_app.elasticsearch.delete(index=index, doc_type=index, id=object_id)


def query_index(index, query, page, per_page):
    if not current_app.elasticsearch:
        return [], 0

    search = current_app.elasticsearch.search(index=index, body={"query": query})

    items = []
    for hit in search["hits"]["hits"]:
        item = hit["_source"]
        item["id"] = hit["_id"]
        items.append(item)

    return items, search["hits"]["total"]