import json
import logging
import requests
from django.template.loader import render_to_string
logger = logging.getLogger(__name__)
[docs]
class ExternalAutocomplete:
"""
This is a helper base class for implementing external
autocomplete classes. <Modelname>ExernalAutocomplete classes
are expected to have a `get_results(self, q)` method that
returns a list of results usable by the autocomplete view.
This base class implements this `get_results` method in a
way that you can inherit from it and just define a list of
`adapters`. Those adapters are then used one by one to
add external autocomplete search results.
"""
session = requests.Session()
adapters = []
[docs]
def get_results(self, q):
results = []
for adapter in self.adapters:
results.extend(adapter.get_results(q, self.session))
return results
[docs]
class ExternalAutocompleteAdapter:
"""
Base class for ExternalAutocompleteAdapters. It provides
the methods used for templating the autocomplete results.
You can pass a `template` name to initialization, which
is then used to style the results.
"""
template = None
def __init__(self, *args, **kwargs):
self.template = kwargs.get("template", None)
[docs]
def default_template(self, result):
return f'{result["label"]} <a href="{result["id"]}">{result["id"]}</a>'
[docs]
def get_result_label(self, result):
if self.template:
return render_to_string(self.template, {"result": result})
return self.default_template(result)
[docs]
class TypeSenseAutocompleteAdapter(ExternalAutocompleteAdapter):
"""
This autocomplete adapters queries typesense collections on a
typesense server. The `collections` variable can either be a
string or a list - if its a string, that collection is queried
directly, if its a list, the adapter uses typesense `multi_search`
endpoint.
"""
collections = None
token = None
server = None
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.collections = kwargs.get("collections", None)
self.token = kwargs.get("token", None)
self.server = kwargs.get("server", None)
[docs]
def default_template(self, result):
return super().default_template(result["document"])
[docs]
def get_results(self, q, session=requests.Session()):
headers = {"X-TYPESENSE-API-KEY": self.token}
res = None
if self.token and self.server:
match self.collections:
# if there is only on collection configured, we hit that collection directly
case str() as collection:
url = f"{self.server}/collections/{collection}/documents/search?q={q}&query_by=description&query_by=label"
res = session.get(url, headers=headers)
# if there are multiple collections configured, we use the `multi_search` endpoint
case list() as collectionlist:
url = f"{self.server}/multi_search?q={q}&query_by=description&query_by=label"
data = {"searches": []}
for collection in collectionlist:
data["searches"].append({"collection": collection})
res = session.post(url, data=json.dumps(data), headers=headers)
case unknown:
logger.error("Don't know what to do with collection %s", unknown)
if res:
data = res.json()
hits = data.get("hits", [])
for result in data.get("results", []):
hits.extend(result["hits"])
return list(filter(bool, map(self.extract, hits)))
return []
[docs]
class LobidAutocompleteAdapter(ExternalAutocompleteAdapter):
"""
This autocomplete adapters queries the lobid autocomplete apis.
See https://lobid.org/gnd/api for details
You can pass a `lobid_params` dict which will then be use as GET
request parameters.
"""
params = {}
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.params = kwargs.get("params", {})
[docs]
def get_results(self, q, session=requests.Session()):
endpoint = "https://lobid.org/gnd/search?"
self.params["q"] = q
res = session.get(endpoint, params=self.params)
if res:
return list(filter(bool, map(self.extract, res.json())))
return []