50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""Interactions with the Meilisearch API for adding and searching cables."""
|
|
from meilisearch import Client
|
|
from meilisearch.task import TaskInfo
|
|
import json
|
|
|
|
DEFAULT_URL = "http://localhost:7700"
|
|
DEFAULT_APIKEY = "fluffybunnyrabbit" # I WOULD RECOMMEND SOMETHING MORE SECURE
|
|
DEFAULT_INDEX = "cables"
|
|
|
|
|
|
class JukeboxSearch:
|
|
"""Class for interacting with the Meilisearch API."""
|
|
def __init__(self, url: str = None, api_key: str = None, index: str = None):
|
|
"""Connect to Meilisearch and perform first-run tasks as necessary.
|
|
|
|
:param url: Address of the Meilisearch server. Defaults to ``http://localhost:7700`` if unspecified.
|
|
:param api_key: API key used to authenticate with Meilisearch. It is highly recommended to set this as something
|
|
secure if you can access this endpoint publicly, but you can ignore this and set Meilisearch's default API key
|
|
to ``fluffybunnyrabbit``.
|
|
:param index: The name of the index to configure. Defaults to ``cables`` if unspecified."""
|
|
# connect to Meilisearch
|
|
url = url or DEFAULT_URL
|
|
api_key = api_key or DEFAULT_APIKEY
|
|
self.index = index or DEFAULT_INDEX
|
|
self.client = Client(url, api_key)
|
|
# create the index if it does not exist already
|
|
if self.client.get_index(self.index) is None:
|
|
self.client.create_index(self.index)
|
|
|
|
def add_document(self, document: dict) -> TaskInfo:
|
|
"""Add a cable to the Meilisearch index.
|
|
|
|
:param document: Dictionary containing all the cable data.
|
|
:returns: A TaskInfo object for the addition of the new document."""
|
|
return self.client.index(self.index).add_documents(document)
|
|
|
|
def add_documents(self, documents: list):
|
|
"""Add a list of cables to the Meilisearch index.
|
|
|
|
:param documents: List of dictionaries containing all the cable data.
|
|
:returns: A TaskInfo object for the last new document."""
|
|
taskinfo = None
|
|
for i in documents:
|
|
taskinfo = self.add_document(i)
|
|
return taskinfo
|
|
|
|
def query(self):
|
|
"""Execute a search query on the Meilisearch index."""
|
|
pass
|