Coverage for ramose/api_manager.py: 97%
191 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-24 21:59 +0000
1# SPDX-FileCopyrightText: 2018-2021 Silvio Peroni <silvio.peroni@unibo.it>
2# SPDX-FileCopyrightText: 2020-2021 Marilena Daquino <marilena.daquino2@unibo.it>
3# SPDX-FileCopyrightText: 2022 Davide Brembilla
4# SPDX-FileCopyrightText: 2024 Ivan Heibi <ivan.heibi2@unibo.it>
5# SPDX-FileCopyrightText: 2025 Sergei Slinkin
6# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it>
7#
8# SPDX-License-Identifier: ISC
10from __future__ import annotations
12import csv
13from collections import OrderedDict
14from importlib import import_module
15from pathlib import Path
16from re import findall, match, sub
17from sys import maxsize, path
18from typing import TYPE_CHECKING, TypedDict
19from urllib.parse import urlsplit
21from ramose._constants import DEFAULT_HTTP_TIMEOUT, FORMAT_PARTS_WITH_MEDIA_TYPE, PARAM_NAME
22from ramose.cache import ResultCache
23from ramose.filters import load_yaml_config
24from ramose.hash_format import parse_auth, parse_custom_params, parse_disable_params, read_spec_file
25from ramose.operation import Operation, OperationConfig, OperationResponse
27if TYPE_CHECKING:
28 import types
30 from ramose.filters import FiltersConfig
33class APIConfig(TypedDict):
34 conf: OrderedDict[str, list[dict[str, str]]]
35 conf_json: list[dict[str, str]]
36 base_url: str
37 tp: str
38 update_endpoint: str
39 website: str
40 sources_map: dict[str, str]
41 disable_params: set[str]
42 auth_required: bool
43 addon: types.ModuleType | None
44 sparql_http_method: str
45 conf_file: str
48class APIManager:
49 # Fixing max size for CSV
50 @staticmethod
51 def __max_size_csv() -> None:
52 max_int = maxsize
53 while True:
54 try:
55 csv.field_size_limit(max_int)
56 break
57 except OverflowError: # pragma: no cover
58 max_int = int(max_int / 10)
60 @staticmethod
61 def _load_addon(addon_name: str, conf_file: str) -> types.ModuleType:
62 addon_path = (Path(conf_file).parent / addon_name).resolve()
63 if addon_path.parent.joinpath(f"{addon_path.name}.py").is_file():
64 path.append(str(addon_path.parent))
65 return import_module(addon_path.name)
66 return import_module(addon_name)
68 @staticmethod
69 def _process_api_metadata(
70 conf_json: list[dict[str, str]],
71 conf_file: str,
72 endpoint_override: str | None,
73 ) -> APIConfig:
74 item = conf_json[0]
75 base_url = item["url"]
76 website = item["base"]
77 website_parsed = urlsplit(website)
78 if not website_parsed.scheme or not website_parsed.netloc:
79 msg = "API #base must be an absolute URL"
80 raise ValueError(msg)
81 tp = endpoint_override or item["endpoint"]
82 update_endpoint = endpoint_override or ""
83 if not endpoint_override and "update_endpoint" in item:
84 update_endpoint = item["update_endpoint"]
85 sources_map: dict[str, str] = {}
86 if "sources" in item:
87 for raw_pair in item["sources"].split(";"):
88 pair = raw_pair.strip()
89 if not pair:
90 continue
91 name, url = pair.split("=", 1)
92 sources_map[name.strip()] = url.strip()
93 disable_params_api = parse_disable_params(item["disable_params"]) if "disable_params" in item else set()
94 auth_required = parse_auth(item["auth"]) if "auth" in item else False
95 addon = APIManager._load_addon(item["addon"], conf_file) if "addon" in item else None
96 sparql_http_method = item["method"].strip().lower() if "method" in item else "get"
98 conf: OrderedDict[str, list[dict[str, str]]] = OrderedDict()
99 for op_item in conf_json[1:]:
100 conf.setdefault(APIManager.nor_api_url(op_item, base_url), []).append(op_item)
102 return {
103 "conf": conf,
104 "conf_json": conf_json,
105 "base_url": base_url,
106 "tp": tp or "",
107 "update_endpoint": update_endpoint,
108 "website": website,
109 "sources_map": sources_map,
110 "disable_params": disable_params_api,
111 "auth_required": auth_required,
112 "addon": addon,
113 "sparql_http_method": sparql_http_method,
114 "conf_file": conf_file,
115 }
117 def __init__( # noqa: PLR0913, PLR0917
118 self,
119 conf_files: list[str],
120 endpoint_override: str | None = None,
121 cache_dir: str | None = None,
122 cache_ttl: int = 86400,
123 retry_attempts: int = 3,
124 retry_wait: float = 0.5,
125 retry_backoff: float = 2.0,
126 sparql_timeout: float = DEFAULT_HTTP_TIMEOUT,
127 ) -> None:
128 """This is the constructor of the APIManager class. It takes in input a list of API configuration files, each
129 defined according to the Hash Format or YAML mirror format, and stores all the operations defined within a
130 dictionary. Optionally, an endpoint_override parameter can be provided to override the SPARQL endpoint defined
131 in the configuration files (useful for staging/production environments).
132 The structure of each item in the dictionary of the operations is defined as follows:
134 {
135 "/api/v1/references/(.+)": {
136 "sparql": "PREFIX ...",
137 "method": "get",
138 ...
139 },
140 ...
141 }
143 In particular, each key in the dictionary identifies the full URL of a particular API operation, and it is
144 used so as to understand with operation should be called once an API call is done. The object associated
145 as value of this key is the transformation of the related operation defined in the input spec file into a
146 dictionary.
148 In addition, it also defines additional structure, such as the functions to be used for interpreting the
149 values returned by a SPARQL query, some operations that can be used for filtering the results, and the
150 HTTP methods to call for making the request to the SPARQL endpoint specified in the configuration file."""
151 APIManager.__max_size_csv()
153 self._cache = ResultCache(cache_dir) if cache_dir else None
154 self._cache_ttl = cache_ttl
155 self._config_cache: dict[str, FiltersConfig] = {}
156 self._retry_attempts = retry_attempts
157 self._retry_wait = retry_wait
158 self._retry_backoff = retry_backoff
159 self._sparql_timeout = sparql_timeout
161 self.all_conf: OrderedDict[str, APIConfig] = OrderedDict()
162 self.base_url: list[str] = []
163 for conf_file in conf_files:
164 conf_json = read_spec_file(conf_file)
165 if not conf_json:
166 continue
167 api_conf = APIManager._process_api_metadata(conf_json, conf_file, endpoint_override)
168 self.base_url.append(api_conf["base_url"])
169 self.all_conf[api_conf["base_url"]] = api_conf
171 self._operation_prefixes = APIManager._build_operation_prefixes(self.all_conf)
173 @staticmethod
174 def _build_operation_prefixes(
175 all_conf: OrderedDict[str, APIConfig],
176 ) -> list[tuple[str, str, dict[str, str]]]:
177 prefixes: list[tuple[str, str, dict[str, str]]] = []
178 for base, api_data in all_conf.items():
179 for items in api_data["conf"].values():
180 for item in items:
181 template = item["url"]
182 brace_pos = template.find("{")
183 if brace_pos != -1:
184 prefixes.append((base + template[:brace_pos], base, item))
185 prefixes.sort(key=lambda entry: len(entry[0]), reverse=True)
186 return prefixes
188 @staticmethod
189 def nor_api_url(i: dict[str, str], b: str = "") -> str:
190 """This method takes an API operation object and an optional base URL (e.g. "/api/v1") as input
191 and returns the URL composed by the base URL plus the API URL normalised according to specific rules. In
192 particular, these normalisation rules takes the operation URL (e.g. "#url /citations/{oci}") and the
193 specification of the shape of all the parameters between brackets in the URL (e.g. "#oci str([0-9]+-[0-9]+)"),
194 and returns a new operation URL where the parameters have been substituted with the regular expressions
195 defining them (e.g. "/citations/([0-9]+-[0-9]+)"). This URL will be used by RAMOSE for matching the
196 particular API calls with the specific operation to execute."""
197 result = i["url"]
199 for term in findall(PARAM_NAME, result):
200 try:
201 t = i[term]
202 except KeyError:
203 t = "str(.+)"
204 result = result.replace(f"{{{term}}}", "{}".format(sub(r"^[^\(]+(\(.+\))$", r"\1", t)))
206 return f"{b}{result}"
208 def best_match(self, u: str, method: str = "get") -> tuple[APIConfig | None, str | None, dict[str, str] | None]:
209 """This method takes an URL of an API call and the HTTP method in input and finds the API operation URL,
210 the related configuration and the operation matching the requested method that best match with the API
211 call, if any. When the path matches but no operation declares the requested method, the first operation
212 for that path is returned so that the 405 check can reject it."""
213 cur_u = sub(r"\?.*$", "", u)
214 requested = method.lower()
215 for base_url in self.all_conf:
216 if u.startswith(base_url):
217 conf = self.all_conf[base_url]
218 for pat, items in conf["conf"].items():
219 if match(f"^{pat}$", cur_u):
220 for op_item in items:
221 if requested in op_item["method"].split():
222 return conf, pat, op_item
223 return conf, pat, items[0]
224 return None, None, None
226 @staticmethod
227 def _parse_format_map(op_conf: dict[str, str]) -> tuple[dict[str, str], dict[str, str]]:
228 op_format_map: dict[str, str] = {}
229 op_media_types: dict[str, str] = {}
230 if "format" not in op_conf:
231 return op_format_map, op_media_types
232 fm_val = op_conf["format"]
233 fm_list = fm_val if isinstance(fm_val, list) else [fm_val]
234 for fm in fm_list:
235 for raw_part in fm.split(";"):
236 part = raw_part.strip()
237 if not part:
238 continue
239 fields = [field.strip() for field in part.split(",")]
240 op_format_map[fields[0]] = fields[1]
241 if len(fields) >= FORMAT_PARTS_WITH_MEDIA_TYPE and fields[2]:
242 op_media_types[fields[0]] = fields[2]
243 return op_format_map, op_media_types
245 def _retry_config_for_operation(self, op_conf: dict[str, str]) -> tuple[int, float, float]:
246 retry_attempts = int(op_conf["retry_attempts"]) if "retry_attempts" in op_conf else self._retry_attempts
247 retry_wait = float(op_conf["retry_wait"]) if "retry_wait" in op_conf else self._retry_wait
248 retry_backoff = float(op_conf["retry_backoff"]) if "retry_backoff" in op_conf else self._retry_backoff
249 return retry_attempts, retry_wait, retry_backoff
251 def _load_config(self, conf: APIConfig, handler: str) -> FiltersConfig:
252 resolved = str((Path(conf["conf_file"]).parent / handler).resolve())
253 if resolved not in self._config_cache:
254 self._config_cache[resolved] = load_yaml_config(resolved)
255 return self._config_cache[resolved]
257 def _resolve_custom_param_configs(
258 self, conf: APIConfig, custom_params_map: dict[str, dict[str, str]]
259 ) -> dict[str, FiltersConfig]:
260 result: dict[str, FiltersConfig] = {}
261 for name, param_conf in custom_params_map.items():
262 handler = param_conf["handler"]
263 if param_conf["phase"] != "preprocess" or not handler.endswith((".yaml", ".yml")):
264 continue
265 result[name] = self._load_config(conf, handler)
266 return result
268 def get_op(self, op_complete_url: str, method: str = "get") -> Operation | OperationResponse:
269 """This method returns a new object of type Operation which represent the operation specified by
270 the input URL (parameter 'op_complete_url)' and the HTTP method. If no operation matches,
271 it returns an error OperationResponse."""
272 url_parsed = urlsplit(op_complete_url)
273 op_url = url_parsed.path
275 conf, op, op_conf = self.best_match(op_url, method)
276 if conf is not None and op is not None and op_conf is not None:
277 custom_params_map = parse_custom_params(op_conf["custom_params"]) if "custom_params" in op_conf else {}
279 api_disabled = conf["disable_params"]
280 op_disabled = parse_disable_params(op_conf["disable_params"]) if "disable_params" in op_conf else set()
281 effective_disabled = api_disabled | op_disabled
283 requires_auth = parse_auth(op_conf["auth"]) if "auth" in op_conf else conf["auth_required"]
285 op_format_map, op_format_media_types = APIManager._parse_format_map(op_conf)
286 retry_attempts, retry_wait, retry_backoff = self._retry_config_for_operation(op_conf)
287 config = OperationConfig(
288 sparql_endpoint=conf["tp"],
289 update_endpoint=conf["update_endpoint"],
290 sparql_http_method=conf["sparql_http_method"],
291 addon=conf["addon"],
292 format_map=op_format_map,
293 format_media_types=op_format_media_types,
294 sources_map=conf["sources_map"],
295 custom_params=custom_params_map,
296 disabled_params=effective_disabled,
297 requires_auth=requires_auth,
298 cache=self._cache,
299 default_cache_ttl=self._cache_ttl,
300 custom_param_configs=self._resolve_custom_param_configs(conf, custom_params_map),
301 public_base_url=conf["website"],
302 retry_attempts=retry_attempts,
303 retry_wait=retry_wait,
304 retry_backoff=retry_backoff,
305 sparql_timeout=float(op_conf["sparql_timeout"])
306 if "sparql_timeout" in op_conf
307 else self._sparql_timeout,
308 )
309 return Operation(op_complete_url, op, op_conf, config)
311 for prefix, base_url, item in self._operation_prefixes:
312 if op_url.startswith(prefix):
313 template = item["url"]
314 full_template = base_url + template
315 param_value = op_url[len(prefix) :]
316 param_names = findall(PARAM_NAME, template)
317 if not param_value:
318 msg = (
319 f"HTTP status code 400: the operation '{full_template}' "
320 f"requires a value for parameter '{param_names[0]}'"
321 )
322 else:
323 msg = (
324 f"HTTP status code 400: the value '{param_value}' is not valid for parameter "
325 f"'{param_names[0]}' in operation '{full_template}'"
326 )
327 if "call" in item:
328 msg += f". Example: {base_url}{item['call']}"
329 return OperationResponse.error(400, msg)
331 return OperationResponse.error(404, "HTTP status code 404: the operation requested does not exist")