Coverage for apis_core/utils/autocomplete.py: 0%
101 statements
« prev ^ index » next coverage.py v7.5.3, created at 2025-12-04 11:32 +0000
1import base64
2import json
3import logging
4from urllib.parse import urlparse
6import httpx
7from django.template.loader import render_to_string
9logger = logging.getLogger(__name__)
12class ExternalAutocomplete:
13 """
14 This is a helper base class for implementing external
15 autocomplete classes. <Modelname>ExernalAutocomplete classes
16 are expected to have a `get_results(self, q)` method that
17 returns a list of results usable by the autocomplete view.
18 This base class implements this `get_results` method in a
19 way that you can inherit from it and just define a list of
20 `adapters`. Those adapters are then used one by one to
21 add external autocomplete search results.
22 """
24 client = httpx.Client()
25 adapters = []
27 def get_results(self, q):
28 results = []
29 for adapter in self.adapters:
30 results.extend(adapter.get_results(q, self.client))
31 return results
34class ExternalAutocompleteAdapter:
35 """
36 Base class for ExternalAutocompleteAdapters. It provides
37 the methods used for templating the autocomplete results.
38 You can pass a `template` name to initialization, which
39 is then used to style the results.
40 """
42 template = None
44 def __init__(self, *args, **kwargs):
45 self.template = kwargs.get("template", None)
46 self.data_mapping = kwargs.get("data_mapping", {})
48 def _nested_get(self, dict_, keys):
49 """
50 Get a nested value from a dict by poviding the
51 path as a list of keys
52 """
53 for index, key in enumerate(keys):
54 if index == len(keys) - 1:
55 return [dict_.get(key, "")]
56 dict_ = dict_.get(key, {})
58 def map_data(self, result) -> dict:
59 """
60 Map data from the result to a a new dict. The new dict
61 is built using the settings in `.data_mapping`, which
62 should be a mapping of keys to be created in the result
63 dict to paths in the `result` item.
64 """
65 data = {}
66 for key, val in self.data_mapping.items():
67 if isinstance(val, str):
68 val = [val]
69 data[key] = self._nested_get(result, val)
70 return data
72 def add_data_to_uri(self, uri, data):
73 b64_data = base64.b64encode(json.dumps(data).encode()).decode("ascii")
74 if urlparse(uri).query:
75 uri += f"&anero_ac_data={b64_data}"
76 else:
77 uri += f"?anero_ac_data={b64_data}"
78 return uri
80 def default_template(self, result):
81 return f'{result["label"]} <a href="{result["id"]}">{result["id"]}</a>'
83 def get_result_label(self, result):
84 if self.template:
85 return render_to_string(self.template, {"result": result})
86 return self.default_template(result)
89class TypeSenseAutocompleteAdapter(ExternalAutocompleteAdapter):
90 """
91 This autocomplete adapters queries typesense collections on a
92 typesense server. The `collections` variable can either be a
93 string or a list - if its a string, that collection is queried
94 directly, if its a list, the adapter uses typesense `multi_search`
95 endpoint.
96 """
98 collections = None
99 token = None
100 server = None
102 def __init__(self, *args, **kwargs):
103 super().__init__(*args, **kwargs)
104 self.collections = kwargs.get("collections", None)
105 self.token = kwargs.get("token", None)
106 self.server = kwargs.get("server", None)
108 def default_template(self, result):
109 return super().default_template(result["document"])
111 def extract(self, res):
112 if res.get("document"):
113 data = self.map_data(res)
114 uri = self.add_data_to_uri(res["document"]["id"], data)
115 return {
116 "id": uri,
117 "text": self.get_result_label(res),
118 "selected_text": self.get_result_label(res),
119 }
120 logger.error(
121 "Could not parse result from typesense collection %s: %s",
122 self.collections,
123 res,
124 )
125 return False
127 def get_results(self, q, client=httpx.Client()):
128 headers = {"X-TYPESENSE-API-KEY": self.token}
129 res = None
130 if self.token and self.server:
131 match self.collections:
132 # if there is only on collection configured, we hit that collection directly
133 case str() as collection:
134 url = f"{self.server}/collections/{collection}/documents/search?q={q}&query_by=description&query_by=label"
135 res = client.get(url, headers=headers)
136 # if there are multiple collections configured, we use the `multi_search` endpoint
137 case list() as collectionlist:
138 url = f"{self.server}/multi_search?q={q}&query_by=description&query_by=label"
139 data = {"searches": []}
140 for collection in collectionlist:
141 data["searches"].append({"collection": collection})
142 res = client.post(url, data=json.dumps(data), headers=headers)
143 case unknown:
144 logger.error("Don't know what to do with collection %s", unknown)
146 if res:
147 data = res.json()
148 hits = data.get("hits", [])
149 for result in data.get("results", []):
150 hits.extend(result["hits"])
151 return list(filter(bool, map(self.extract, hits)))
152 return []
155class LobidAutocompleteAdapter(ExternalAutocompleteAdapter):
156 """
157 This autocomplete adapters queries the lobid autocomplete apis.
158 See https://lobid.org/gnd/api for details
159 You can pass a `lobid_params` dict which will then be use as GET
160 request parameters.
161 """
163 params = {}
165 def __init__(self, *args, **kwargs):
166 super().__init__(*args, **kwargs)
167 self.params = kwargs.get("params", {})
169 def extract(self, res):
170 data = self.map_data(res)
171 uri = self.add_data_to_uri(res["id"], data)
172 return {
173 "id": uri,
174 "text": self.get_result_label(res),
175 "selected_text": self.get_result_label(res),
176 }
178 def get_results(self, q, client=httpx.Client()):
179 endpoint = "https://lobid.org/gnd/search?"
180 self.params["q"] = q
181 res = client.get(endpoint, params=self.params)
182 if res:
183 return list(filter(bool, map(self.extract, res.json())))
184 return []