Coverage for ramose/operation.py: 96%
1110 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 time
13from concurrent.futures import ThreadPoolExecutor
14from csv import DictReader, reader, writer
15from dataclasses import dataclass
16from dataclasses import field as dataclass_field
17from http import HTTPStatus
18from io import StringIO
19from itertools import product
20from json import dumps
21from math import ceil
22from operator import eq, gt, itemgetter, lt
23from re import error as regex_error
24from re import findall, fullmatch, match, search, sub
25from typing import TYPE_CHECKING, NoReturn, TypedDict, cast
26from urllib.parse import parse_qs, quote, urlsplit
28from requests.exceptions import RequestException
29from requests.exceptions import Timeout as RequestsTimeout
31try:
32 from pysparql_anything import SparqlAnything # pyright: ignore[reportMissingImports, reportAttributeAccessIssue]
33except ImportError:
34 SparqlAnything = None
36from ramose._constants import (
37 DEFAULT_HTTP_TIMEOUT,
38 FIELD_TYPE_RE,
39 _http_session,
40 backend_auth,
41 media_type_for_format,
42)
43from ramose.datatype import DataType
44from ramose.escaping import InvalidParameterValueError, escape_iri, escape_literal
45from ramose.filters import apply_filters
46from ramose.paging import PaginationInfo, build_link_header, build_pagination_info
48if TYPE_CHECKING:
49 import types
50 from collections.abc import Callable, Mapping
51 from typing import Protocol
53 from requests import Response
55 from ramose.cache import ResultCache
56 from ramose.filters import FiltersConfig
58 class SparqlAnythingEngine(Protocol):
59 def select(self, output_type: type[object], **kwargs: object) -> object: ...
62_WRITE_METHODS = frozenset({"post", "put", "delete"})
63_RETRYABLE_STATUS_CODES = frozenset(
64 {
65 HTTPStatus.REQUEST_TIMEOUT,
66 HTTPStatus.TOO_MANY_REQUESTS,
67 HTTPStatus.INTERNAL_SERVER_ERROR,
68 HTTPStatus.BAD_GATEWAY,
69 HTTPStatus.SERVICE_UNAVAILABLE,
70 HTTPStatus.GATEWAY_TIMEOUT,
71 }
72)
73_SPARQL_ANYTHING_HTTP_STATUS_RE = r"\bHTTP/\d(?:\.\d)?\s+(\d{3})\b"
74_SPARQL_ANYTHING_TIMEOUT_MARKERS = frozenset(
75 {
76 "sockettimeoutexception",
77 "timed out",
78 }
79)
80_SPARQL_ANYTHING_NETWORK_MARKERS = frozenset(
81 {
82 "httphostconnectexception",
83 "connectexception",
84 "noroutetohostexception",
85 "unknownhostexception",
86 "connection refused",
87 }
88)
89_UNPROCESSABLE_CONTENT = 422
90_JSON_TRANSFORM_RE = r'^(?P<op_type>array|dict)\((?P<separator>"[^"]+"),(?P<entries>[^)]+)\)$'
91_DICT_TRANSFORM_MIN_FIELD_COUNT = 2
92ResultRow = list[str]
93ResultTable = list[ResultRow]
96@dataclass(frozen=True)
97class ConversionResult:
98 status_code: int
99 body: str
102@dataclass(frozen=True)
103class OperationResponse:
104 status_code: int
105 body: str
106 content_type: str
107 headers: dict[str, str]
108 is_error_message: bool = False
110 @classmethod
111 def error(cls, status_code: int, body: str) -> OperationResponse:
112 return cls(status_code, body, "text/plain", {}, is_error_message=True)
115class CachedPagination(TypedDict):
116 page: int
117 page_size: int
118 total_items: int
121class CachedResult(TypedDict):
122 rows: ResultTable
123 pagination: CachedPagination | None
126class HttpError(Exception):
127 def __init__(self, status_code: int, message: str) -> None:
128 super().__init__(message)
129 self.status_code = status_code
132@dataclass
133class OperationConfig:
134 sparql_endpoint: str = ""
135 update_endpoint: str = ""
136 sparql_http_method: str = "get"
137 addon: types.ModuleType | None = None
138 format_map: dict = dataclass_field(default_factory=dict)
139 format_media_types: dict = dataclass_field(default_factory=dict)
140 sources_map: dict = dataclass_field(default_factory=dict)
141 custom_params: dict = dataclass_field(default_factory=dict)
142 disabled_params: set = dataclass_field(default_factory=set)
143 requires_auth: bool = False
144 cache: ResultCache | None = None
145 default_cache_ttl: int = 86400
146 custom_param_configs: dict[str, FiltersConfig] = dataclass_field(default_factory=dict)
147 public_base_url: str = ""
148 retry_attempts: int = 3
149 retry_wait: float = 0.5
150 retry_backoff: float = 2.0
151 sparql_timeout: float = DEFAULT_HTTP_TIMEOUT
153 def __post_init__(self) -> None:
154 if self.retry_attempts < 1:
155 msg = "retry_attempts must be >= 1"
156 raise ValueError(msg)
157 if self.sparql_timeout <= 0:
158 msg = "sparql_timeout must be > 0"
159 raise ValueError(msg)
160 if self.retry_wait < 0:
161 msg = "retry_wait must be >= 0"
162 raise ValueError(msg)
163 if self.retry_backoff < 1:
164 msg = "retry_backoff must be >= 1"
165 raise ValueError(msg)
168class Operation:
169 def __init__(
170 self,
171 op_complete_url: str,
172 op_key: str,
173 op_item: dict[str, str],
174 config: OperationConfig | None = None,
175 ) -> None:
176 if config is None:
177 config = OperationConfig()
178 self.url_parsed = urlsplit(op_complete_url)
179 self.op_url = self.url_parsed.path
180 self.op = op_key
181 self.i = op_item
182 self.tp = config.sparql_endpoint
183 self.update_endpoint = config.update_endpoint
184 self.sparql_http_method = config.sparql_http_method
185 self.addon = config.addon
186 self.format = config.format_map
187 self.format_media_types = config.format_media_types
188 self.sources_map = config.sources_map
189 self.custom_params = config.custom_params
190 self.disabled_params = config.disabled_params
191 self.requires_auth = config.requires_auth
192 self._sa_engine = None
193 self._cache = config.cache
194 self._default_cache_ttl = config.default_cache_ttl
195 self.custom_param_configs = config.custom_param_configs
196 self.public_base_url = config.public_base_url
197 self.retry_attempts = config.retry_attempts
198 self.retry_wait = config.retry_wait
199 self.retry_backoff = config.retry_backoff
200 self.sparql_timeout = config.sparql_timeout
201 self.pagination_info: PaginationInfo | None = None
203 self.operation = {"=": eq, "<": lt, ">": gt}
205 self.dt = DataType()
207 def _public_request_url(self, request_url: str) -> str:
208 parsed = urlsplit(request_url)
209 if parsed.scheme and parsed.netloc:
210 return request_url
211 return f"{self.public_base_url.rstrip('/')}/{request_url.lstrip('/')}"
213 def _converter_request_url(self) -> str:
214 if self.pagination_info is not None:
215 return self._public_request_url(self.pagination_info.self_url)
216 if self.url_parsed.query:
217 return self._public_request_url(f"{self.op_url}?{self.url_parsed.query}")
218 return self._public_request_url(self.op_url)
220 def _is_builtin_param_active(self, name: str) -> bool:
221 return name not in self.disabled_params and name not in self.custom_params
223 @staticmethod
224 def _raise_unprocessable(message: str) -> NoReturn:
225 raise HttpError(_UNPROCESSABLE_CONTENT, f"HTTP status code {_UNPROCESSABLE_CONTENT}: {message}")
227 @staticmethod
228 def _parse_positive_int_param(params: dict[str, list[str]], name: str) -> int:
229 raw_value = params[name][0]
230 try:
231 value = int(raw_value)
232 except ValueError:
233 Operation._raise_unprocessable(f"{name} must be an integer, got {raw_value!r}")
234 if value < 1:
235 Operation._raise_unprocessable(f"{name} must be >= 1, got {value}")
236 return value
238 @staticmethod
239 def _validate_page_range(page: int, total_items: int, total_pages: int) -> None:
240 if total_items and page > total_pages:
241 Operation._raise_unprocessable(f"page {page} exceeds total pages {total_pages}")
243 @staticmethod
244 def get_content_type(ct: str) -> str:
245 """It returns the mime type of a given textual representation of a format, being it either
246 'csv' or 'json."""
247 content_type = ct
249 if ct == "csv":
250 content_type = "text/csv"
251 elif ct == "json":
252 content_type = "application/json"
254 return content_type
256 def _convert(self, s: str, fmt: str) -> ConversionResult:
257 converter_func = getattr(self.addon, self.format[fmt])
258 converted = converter_func(s, request_url=self._converter_request_url(), base_url=self.public_base_url)
259 if isinstance(converted, ConversionResult):
260 return converted
261 return ConversionResult(status_code=HTTPStatus.OK, body=converted)
263 def _resolve_format(self, s: str, query_string: dict[str, list[str]]) -> tuple[ConversionResult, str] | None:
264 if "format" in query_string and self._is_builtin_param_active("format"):
265 for req_format in query_string["format"]:
266 if req_format in self.format:
267 return self._convert(s, req_format), self._media_type_for_format(req_format)
268 elif "default_format" in self.i:
269 default_fmt = self.i["default_format"].strip()
270 if default_fmt in self.format:
271 return self._convert(s, default_fmt), self._media_type_for_format(default_fmt)
272 return None
274 def _media_type_for_format(self, fmt: str) -> str:
275 if fmt in self.format_media_types:
276 return self.format_media_types[fmt]
277 return Operation.get_content_type(fmt)
279 def _validate_format_values(self, formats: list[str]) -> None:
280 supported_formats = {"csv", "json", *self.format}
281 for req_format in formats:
282 if req_format not in supported_formats:
283 Operation._raise_unprocessable(f"unsupported format {req_format!r}")
285 def media_type_to_format(self) -> dict[str, str]:
286 if not self._is_builtin_param_active("format"):
287 return {}
288 default_token = self.i["default_format"].strip() if "default_format" in self.i else "json"
289 media_type_to_token: dict[str, str] = {}
290 for token in [default_token, "json", "csv", *self.format]:
291 if token in self.format_media_types:
292 media_type = self.format_media_types[token]
293 else:
294 media_type = media_type_for_format(token)
295 if media_type is not None:
296 media_type_to_token.setdefault(media_type, token)
297 return media_type_to_token
299 def _format_response(
300 self, s: str, query_string: dict[str, list[str]], c_type: str = "text/csv"
301 ) -> tuple[ConversionResult, str]:
302 content_type = Operation.get_content_type(c_type)
304 if "format" in query_string and self._is_builtin_param_active("format"):
305 self._validate_format_values(query_string["format"])
307 resolved = self._resolve_format(s, query_string)
308 if resolved is not None:
309 return resolved
311 if "format" in query_string and self._is_builtin_param_active("format"):
312 for req_format in query_string["format"]:
313 content_type = Operation.get_content_type(req_format)
314 elif "default_format" in self.i:
315 content_type = Operation.get_content_type(self.i["default_format"].strip())
317 if content_type not in ("text/csv", "application/json"):
318 content_type = "text/csv"
320 if "application/json" in content_type:
321 with StringIO(s) as f:
322 r = [dict(i) for i in DictReader(f)]
324 if self._is_builtin_param_active("json"):
325 r = Operation.structured(query_string, r) # type: ignore[arg-type]
327 return ConversionResult(HTTPStatus.OK, dumps(r, ensure_ascii=False, indent=4)), content_type
328 else:
329 return ConversionResult(HTTPStatus.OK, s), content_type
331 def conv(self, s: str, query_string: dict[str, list[str]], c_type: str = "text/csv") -> OperationResponse:
332 """This method takes a string representing a CSV document and converts it in the requested format according
333 to what content type is specified as input."""
334 formatted, content_type = self._format_response(s, query_string, c_type)
335 return OperationResponse(formatted.status_code, formatted.body, content_type, {})
337 @staticmethod
338 def pv(i: int | tuple[object, str], r: list[tuple[object, str]] | None = None) -> str:
339 """This method returns the plain value of a particular item 'i' of the result returned by the SPARQL query.
341 In case 'r' is specified (i.e. a row containing a set of results), then 'i' must be the index of the item
342 within that row."""
343 if r is None:
344 return i[1] # type: ignore[index]
345 return Operation.pv(r[i]) # type: ignore[index]
347 @staticmethod
348 def tv(i: int | tuple[object, str], r: list[tuple[object, str]] | None = None) -> object:
349 """This method returns the typed value of a particular item 'i' of the result returned by the SPARQL query.
350 The type associated to that value is actually specified by means of the particular configuration provided
351 in the specification file of the API - field 'field_type'.
353 In case 'r' is specified (i.e. a row containing a set of results), then 'i' must be the index of the item
354 within that row."""
355 if r is None:
356 return i[0] # type: ignore[index]
357 return Operation.tv(r[i]) # type: ignore[index]
359 @staticmethod
360 def do_overlap(r1: tuple[int, int], r2: tuple[int, int]) -> bool:
361 """This method returns a boolean that says if the two ranges (i.e. two pairs of integers) passed as inputs
362 actually overlap one with the other."""
363 r1_s, r1_e = r1
364 r2_s, r2_e = r2
366 return r1_s <= r2_s <= r1_e or r2_s <= r1_s <= r2_e
368 @staticmethod
369 def get_item_in_dict(
370 d_or_l: dict[str, object] | list[dict[str, object]], key_list: list[str], prev: list[object] | None = None
371 ) -> list[object]:
372 """This method takes as input a dictionary or a list of dictionaries and browses it until the value
373 specified following the chain indicated in 'key_list' is not found. It returns a list of all the
374 values that matched with such search."""
375 res = [] if prev is None else prev.copy()
377 d_list = [d_or_l] if isinstance(d_or_l, dict) else d_or_l
379 for d in d_list:
380 key_list_len = len(key_list)
382 if key_list_len >= 1:
383 key = key_list[0]
384 if key in d:
385 if key_list_len == 1:
386 res.append(d[key])
387 else:
388 res = Operation.get_item_in_dict(d[key], key_list[1:], res) # type: ignore[arg-type]
390 return res
392 @staticmethod
393 def add_item_in_dict(
394 d_or_l: dict[str, object] | list[dict[str, object]], key_list: list[str], item: object, idx: int
395 ) -> None:
396 """This method takes as input a dictionary or a list of dictionaries, browses it until the value
397 specified following the chain indicated in 'key_list' is not found, and then substitutes it with 'item'.
398 In case the final object retrieved is a list, it selects the object in position 'idx' before the
399 substitution."""
400 key_list_len = len(key_list)
402 if key_list_len >= 1:
403 key = key_list[0]
405 if isinstance(d_or_l, list):
406 if key_list_len == 1:
407 d_or_l[idx][key] = item
408 else:
409 for i in d_or_l:
410 Operation.add_item_in_dict(i, key_list, item, idx)
411 elif key in d_or_l:
412 if key_list_len == 1:
413 d_or_l[key] = item
414 else:
415 Operation.add_item_in_dict(d_or_l[key], key_list[1:], item, idx) # type: ignore[arg-type]
417 @staticmethod
418 def _apply_array_transform(row: dict[str, object], keys: list[str], separator: str, v_list: list[object]) -> None:
419 for idx, v in enumerate(v_list):
420 if isinstance(v, str):
421 Operation.add_item_in_dict(row, keys, v.split(separator) if v != "" else [], idx)
423 @staticmethod
424 def _apply_dict_transform(
425 row: dict[str, object], keys: list[str], separator: str, entries: list[str], v_list: list[object]
426 ) -> None:
427 new_fields = entries[1:]
428 new_fields_max_split = len(new_fields) - 1
429 for idx, v in enumerate(v_list):
430 if isinstance(v, str):
431 new_values = v.split(separator, new_fields_max_split)
432 Operation.add_item_in_dict(
433 row,
434 keys,
435 dict(zip(new_fields, new_values, strict=False)) if v != "" else {},
436 idx,
437 )
438 elif isinstance(v, list):
439 new_list = [dict(zip(new_fields, i.split(separator, new_fields_max_split), strict=False)) for i in v]
440 Operation.add_item_in_dict(row, keys, new_list, idx)
442 @staticmethod
443 def structured(params: dict[str, list[str]], json_table: list[dict[str, object]]) -> list[dict[str, object]]:
444 """This method checks if there are particular transformation rules specified in 'params' for a JSON output,
445 and convert each row of the input table ('json_table') according to these rules.
446 There are two specific rules that can be applied:
448 1. array("<separator>",<field>): it converts the string value associated to the field name '<field>' into
449 an array by splitting the various textual parts by means of '<separator>'. For instance, consider the
450 following JSON structure:
452 [
453 { "names": "Doe, John; Doe, Jane" },
454 { "names": "Doe, John; Smith, John" }
455 ]
457 Executing the rule 'array("; ",names)' returns the following new JSON structure:
459 [
460 { "names": [ "Doe, John", "Doe, Jane" ],
461 { "names": [ "Doe, John", "Smith, John" ]
462 ]
464 2. dict("separator",<field>,<new_field_1>,<new_field_2>,...): it converts the string value associated to
465 the field name '<field>' into an dictionary by splitting the various textual parts by means of
466 '<separator>' and by associating the new fields '<new_field_1>', '<new_field_2>', etc., to these new
467 parts. For instance, consider the following JSON structure:
469 [
470 { "name": "Doe, John" },
471 { "name": "Smith, John" }
472 ]
474 Executing the rule 'array(", ",name,family_name,given_name)' returns the following new JSON structure:
476 [
477 { "name": { "family_name": "Doe", "given_name: "John" } },
478 { "name": { "family_name": "Smith", "given_name: "John" } }
479 ]
481 Each of the specified rules is applied in order, and it works on the JSON structure returned after
482 the execution of the previous rule."""
483 if "json" in params:
484 fields = params["json"]
485 for field in fields:
486 op_match = fullmatch(_JSON_TRANSFORM_RE, field)
487 if op_match is None:
488 Operation._raise_unprocessable(f"invalid json transform {field!r}")
489 op_type = op_match.group("op_type")
490 quoted_separator = op_match.group("separator")
491 entries = [i.strip() for i in op_match.group("entries").split(",")]
492 separator = quoted_separator[1:-1]
493 if op_type == "array" and len(entries) != 1:
494 Operation._raise_unprocessable(f"json array transform expects one field, got {field!r}")
495 if op_type == "dict" and len(entries) < _DICT_TRANSFORM_MIN_FIELD_COUNT:
496 Operation._raise_unprocessable(f"json dict transform expects output fields, got {field!r}")
497 keys = entries[0].split(".")
499 for row in json_table:
500 v_list = Operation.get_item_in_dict(row, keys)
501 if op_type == "array":
502 Operation._apply_array_transform(row, keys, separator, v_list)
503 elif op_type == "dict":
504 Operation._apply_dict_transform(row, keys, separator, entries, v_list)
506 return json_table
508 def preprocess(
509 self, par_dict: dict[str, object], op_item: dict[str, str], addon: types.ModuleType
510 ) -> dict[str, object]:
511 """This method takes the a dictionary of parameters with the current typed values associated to them and
512 the item of the API specification defining the behaviour of that operation, and preprocesses the parameters
513 according to the functions specified in the '#preprocess' field (e.g. "#preprocess lower(doi)"), which is
514 applied to the specified parameters as input of the function in consideration (e.g.
515 "/api/v1/citations/10.1108/jd-12-2013-0166", converting the DOI in lowercase).
517 It is possible to run multiple functions sequentially by concatenating them with "-->" in the API
518 specification document. In this case the output of the function f_i will becomes the input operation URL
519 of the function f_i+1.
521 Finally, it is worth mentioning that all the functions specified in the "#preprocess" field must return
522 a tuple of values defining how the particular value passed in the dictionary must be changed."""
523 result = par_dict
525 if "preprocess" in op_item:
526 for pre in [sub(r"\s+", "", i) for i in op_item["preprocess"].split(" --> ")]:
527 func_name = sub(r"^([^\(\)]+)\(.+$", r"\1", pre).strip()
528 params_name = sub(r"^.+\(([^\(\)]+)\).*", r"\1", pre).split(",")
530 param_list = tuple(result[param_name] for param_name in params_name)
532 # run function
533 func = getattr(addon, func_name)
534 res = func(*param_list)
536 # substitute res to the current parameter in result
537 for idx, val in enumerate(res):
538 result[params_name[idx]] = val
540 return result
542 def postprocess(
543 self, res: list[list[str] | list[tuple[object, str]]], op_item: dict[str, str], addon: types.ModuleType
544 ) -> list[list[str] | list[tuple[object, str]]]:
545 """This method takes the result table returned by running the SPARQL query in an API operation (specified
546 as input) and change some of such results according to the functions specified in the '#postprocess'
547 field (e.g. "#postprocess remove_date("2018")"). These functions can take parameters as input, while the first
548 unspecified parameters will be always the result table. It is worth mentioning that this result table (i.e.
549 a list of tuples) actually contains, in each cell, a tuple defining the plain value as well as the typed
550 value for enabling better comparisons and operations if needed. An example of this table of result is shown as
551 follows:
553 [
554 ("id", "date"),
555 ("my_id_1", "my_id_1"), (datetime(2018, 3, 2), "2018-03-02"),
556 ...
557 ]
559 Note that the typed value and the plain value of each cell can be selected by using the methods "tv" and "pv"
560 respectively. In addition, it is possible to run multiple functions sequentially by concatenating them
561 with "-->" in the API specification document. In this case the output of the function f_i will becomes
562 the input result table of the function f_i+1."""
563 result = res
565 if "postprocess" in op_item:
566 for post in [i.strip() for i in op_item["postprocess"].split(" --> ")]:
567 func_name = sub(r"^([^\(\)]+)\(.+$", r"\1", post).strip()
568 param_str = sub(r"^.+\(([^\(\)]*)\).*", r"\1", post)
569 params_values = () if param_str == "" else next(reader(param_str.splitlines(), skipinitialspace=True))
571 func = getattr(addon, func_name)
572 func_params = (result, *tuple(params_values))
573 result, do_type_fields = func(*func_params)
574 if do_type_fields:
575 result = self.type_fields(result, op_item)
577 return result
579 @staticmethod
580 def _apply_require(
581 header: list[str], result: list[list[tuple[object, str]]], fields: list[str]
582 ) -> list[list[tuple[object, str]]]:
583 """Exclude rows with empty values in the specified fields."""
584 for field in fields:
585 if field not in header:
586 Operation._raise_unprocessable(f"require field {field!r} is not in the result header")
587 field_idx = header.index(field)
588 result = [row for row in result if Operation.pv(field_idx, row) not in (None, "")]
589 return result
591 def _apply_filter(
592 self, header: list[str], result: list[list[tuple[object, str]]], fields: list[str]
593 ) -> list[list[tuple[object, str]]]:
594 """Filter rows by comparison operators or regex patterns."""
595 for field in fields:
596 if ":" not in field:
597 Operation._raise_unprocessable(f"filter must use field:value syntax, got {field!r}")
598 field_name, field_value = field.split(":", 1)
599 if field_name not in header:
600 Operation._raise_unprocessable(f"filter field {field_name!r} is not in the result header")
601 if field_value == "":
602 Operation._raise_unprocessable(f"filter value for field {field_name!r} must not be empty")
603 field_idx = header.index(field_name)
604 if not result:
605 continue
606 flag = field_value[0]
607 if flag in ("<", ">", "="):
608 value = field_value[1:].lower()
609 if value == "":
610 Operation._raise_unprocessable(
611 f"filter comparison value for field {field_name!r} must not be empty"
612 )
613 try:
614 typed_value = self.dt.get_func(type(Operation.tv(field_idx, result[0])).__name__)(value)
615 except ValueError:
616 Operation._raise_unprocessable(f"filter value {value!r} is invalid for field {field_name!r}")
617 result = [
618 row
619 for row in result
620 if self.operation[flag](
621 Operation.tv(field_idx, row),
622 typed_value,
623 )
624 ]
625 else:
626 pattern = field_value.lower()
627 try:
628 result = [row for row in result if search(pattern, Operation.pv(field_idx, row).lower())]
629 except regex_error as exc:
630 Operation._raise_unprocessable(f"filter regex for field {field_name!r} is invalid: {exc}")
631 return result
633 @staticmethod
634 def _apply_sort(
635 header: list[str], result: list[list[tuple[object, str]]], fields: list[str]
636 ) -> list[list[tuple[object, str]]]:
637 """Sort rows by the specified fields and directions."""
638 for field in sorted(fields, reverse=True):
639 order_match = fullmatch(r"(?P<direction>desc|asc)\((?P<field_name>[^()]+)\)", field)
640 if order_match:
641 direction = order_match.group("direction")
642 field_name = order_match.group("field_name")
643 else:
644 if "(" in field or ")" in field:
645 Operation._raise_unprocessable(f"invalid sort expression {field!r}")
646 direction, field_name = "asc", field
647 if field_name not in header:
648 Operation._raise_unprocessable(f"sort field {field_name!r} is not in the result header")
649 field_idx = header.index(field_name)
650 result = sorted(result, key=itemgetter(field_idx), reverse=(direction == "desc"))
651 return result
653 def handling_params(
654 self, params: dict[str, list[str]], table: list[list[str] | list[tuple[object, str]]]
655 ) -> list[list[str] | list[tuple[object, str]]]:
656 """This method is used for filtering the results that are returned after the post-processing
657 phase. In particular, it is possible to:
659 1. [require=<field_name>] exclude all the rows that have an empty value in the field specified - e.g. the
660 "require=doi" remove all the rows that do not have any string specified in the "doi" field;
662 2. [filter=<field_name>:<operator><value>] consider only the rows where the string in the input field
663 is compliant with the value specified. If no operation is specified, the value is interpreted as a
664 regular expression, otherwise it is compared according to the particular type associated to that field.
665 Possible operators are "=", "<", and ">" - e.g. "filter=title:semantics?" returns all the rows that contain
666 the string "semantic" or "semantics" in the field title, while "filter=date:>2016-05" returns all the rows
667 that have a date greater than May 2016;
669 3. [sort=<order>(<field_name>)] sort all the results according to the value and type of the particular
670 field specified in input. It is possible to sort the rows either in ascending ("asc") or descending
671 ("desc") order - e.g. "sort=desc(date)" sort all the rows according to the value specified in the
672 field "date" in descending order.
674 Note that these filtering operations are applied in the order presented above - first the "require", then
675 the "filter", and finally the "sort". It is possible to specify one or more filtering operation of the
676 same kind (e.g. "require=doi&require=title").
677 """
678 header = table[0]
679 result = table[1:]
681 overridden = set(self.custom_params) | self.disabled_params
683 if ("exclude" in params or "require" in params) and "require" not in overridden and "exclude" not in overridden:
684 fields = params["exclude"] if "exclude" in params else params["require"]
685 result = self._apply_require(header, result, fields) # type: ignore[arg-type]
687 if "filter" in params and "filter" not in overridden:
688 result = self._apply_filter(header, result, params["filter"]) # type: ignore[arg-type]
690 if "sort" in params and "sort" not in overridden:
691 result = self._apply_sort(header, result, params["sort"]) # type: ignore[arg-type]
693 return [header, *result]
695 def type_fields(
696 self, res: list[list[str] | list[tuple[object, str]] | list[str | object]], op_item: dict[str, str]
697 ) -> list[list[str] | list[tuple[object, str]]]:
698 """It creates a version of the results 'res' that adds, to each value of the fields, the same value interpreted
699 with the type specified in the specification file (field 'field_type'). Note that 'str' is used as default in
700 case no further specifications are provided."""
701 result = []
702 cast_func = {}
703 header = res[0]
704 for heading in header:
705 cast_func[heading] = DataType.str
707 if "field_type" in op_item:
708 for f, p in findall(FIELD_TYPE_RE, op_item["field_type"]):
709 cast_func[p] = self.dt.get_func(f)
711 for row in res[1:]:
712 new_row = []
713 for idx, heading in enumerate(header):
714 cur_value = row[idx]
715 if isinstance(cur_value, tuple):
716 cur_value = cur_value[1]
717 new_row.append((cast_func[heading](cur_value), cur_value))
718 result.append(new_row)
720 return [header, *result] # type: ignore[return-value]
722 def remove_types(self, res: list[list[str] | list[tuple[object, str]]]) -> ResultTable:
723 """This method takes the results 'res' that include also the typed value and returns a version of such
724 results without the types that is ready to be stored on the file system."""
725 result: ResultTable = [cast("list[str]", res[0])]
726 result.extend([Operation.pv(idx, row) for idx in range(len(row))] for row in res[1:]) # type: ignore[arg-type]
727 return result
729 @staticmethod
730 def _is_directive(line: str) -> bool:
731 return line.strip().startswith("@@")
733 @staticmethod
734 def _parse_directive_args(
735 tokens: list[str], param_names: list[str], defaults: dict[str, str] | None = None
736 ) -> dict[str, str]:
737 defaults = defaults or {}
738 all_names = set(param_names) | set(defaults)
739 result = {}
740 positional_index = 0
741 seen_keyword = False
743 for token in tokens:
744 if "=" in token:
745 key, value = token.split("=", 1)
746 if key in all_names:
747 if key in result:
748 msg = f"Duplicate parameter {key!r}"
749 raise ValueError(msg)
750 seen_keyword = True
751 result[key] = value
752 continue
753 if seen_keyword:
754 msg = f"Positional argument {token!r} cannot follow keyword argument"
755 raise ValueError(msg)
756 if positional_index >= len(param_names):
757 msg = f"Unexpected argument {token!r}"
758 raise ValueError(msg)
759 result[param_names[positional_index]] = token
760 positional_index += 1
762 for name, default in defaults.items():
763 if name not in result:
764 result[name] = default
766 missing = [name for name in param_names if name not in result]
767 if missing:
768 msg = f"Missing required parameter(s): {', '.join(missing)}"
769 raise ValueError(msg)
771 return result
773 def _handle_directive_with(self, parts: list[str]) -> tuple[str | None, str, None]:
774 args = Operation._parse_directive_args(
775 parts[1:],
776 ["source"],
777 defaults={"source": "", "endpoint": "", "engine": "sparql"},
778 )
779 name = args["source"].strip()
780 endpoint = args["endpoint"].strip()
781 engine = args["engine"].strip().lower()
782 if engine not in {"sparql", "sparql-anything"}:
783 msg = f"Unknown engine '{args['engine']}' in @@with"
784 raise ValueError(msg)
785 if name and endpoint:
786 msg = "@@with cannot combine source and endpoint"
787 raise ValueError(msg)
788 endpoint_was_given = any(token.startswith("endpoint=") for token in parts[1:])
789 if endpoint_was_given and not endpoint:
790 msg = "@@with endpoint cannot be empty"
791 raise ValueError(msg)
792 if engine == "sparql" and not name and not endpoint:
793 msg = "@@with source or endpoint is required when engine=sparql"
794 raise ValueError(msg)
795 if name and name not in self.sources_map:
796 msg = f"Unknown source '{name}' in @@with; declare it in #sources."
797 raise ValueError(msg)
798 if endpoint:
799 return endpoint, engine, None
800 if name:
801 return self.sources_map[name], engine, None
802 return None, engine, None
804 @staticmethod
805 def _handle_directive_join(parts: list[str]) -> tuple[None, None, tuple[str, str, str, str]]:
806 args = Operation._parse_directive_args(parts[1:], ["left_var", "right_var"], defaults={"type": "inner"})
807 return None, None, ("JOIN", args["left_var"], args["right_var"], args["type"].lower())
809 @staticmethod
810 def _handle_directive_values(parts: list[str]) -> tuple[None, None, tuple[str, list[str], int | None, int]]:
811 variables = [token for token in parts[1:] if "=" not in token]
812 if not variables:
813 msg = "@@values needs at least one variable"
814 raise ValueError(msg)
815 batch_size: int | None = None
816 workers = 1
817 seen_keys: set[str] = set()
818 for token in parts[1:]:
819 if "=" not in token:
820 continue
821 key, raw_value = token.split("=", 1)
822 if key not in ("batch_size", "workers") or key in seen_keys:
823 msg = f"Unexpected argument {token!r}"
824 raise ValueError(msg)
825 seen_keys.add(key)
826 if not raw_value:
827 continue
828 value = int(raw_value)
829 if value < 1:
830 msg = f"{key} must be >= 1, got {value}"
831 raise ValueError(msg)
832 if key == "batch_size":
833 batch_size = value
834 else:
835 workers = value
836 return None, None, ("VALUES_INJECT", variables, batch_size, workers)
838 @staticmethod
839 def _handle_directive_page(parts: list[str]) -> tuple[None, None, tuple[str, str, str, str]]:
840 args = Operation._parse_directive_args(parts[1:], ["variable"], defaults={"default_size": "", "max_size": ""})
841 var_name = args["variable"]
842 if not var_name.startswith("?"):
843 msg = f"@@page variable must start with '?', got {var_name!r}"
844 raise ValueError(msg)
845 return None, None, ("PAGE", var_name, args["default_size"], args["max_size"])
847 def _process_directive(
848 self, line: str, directive_handlers: dict[str, Callable[..., object]]
849 ) -> tuple[str | None, str | None, tuple[str, ...] | None]:
850 body = line.strip()[2:].strip()
851 parts = body.split()
852 cmd = parts[0].lower()
854 handler = directive_handlers.get(cmd)
855 if handler is None:
856 msg = f"Unknown directive @@{cmd}"
857 raise ValueError(msg)
859 return handler(parts) # type: ignore[return-value]
861 @staticmethod
862 def _update_directive_state(
863 current_endpoint: str,
864 current_engine: str,
865 new_endpoint: str | None,
866 new_engine: str | None,
867 ) -> tuple[str, str]:
868 if new_endpoint is not None:
869 current_endpoint = new_endpoint
870 if new_engine is not None:
871 current_engine = new_engine
872 return current_endpoint, current_engine
874 def _parse_steps(self, text: str, default_endpoint: str, params: dict[str, object]) -> list[tuple[str, ...]]:
875 """
876 Returns a list of steps:
877 - ("QUERY", endpoint_url, engine, query_text)
878 - ("JOIN", left_var, right_var, how) # how in {"inner","left"}
879 - ("REMOVE", [vars])
880 - ("VALUES_INJECT", [vars], batch_size, workers) # @@values ?var1 ... [batch_size=N] [workers=N]
881 - ("PAGE", var_name, default_size, max_size) # @@page ?var [default_size=N] [max_size=M]
882 """
883 for p, v in params.items():
884 text = text.replace(f"[[{p}]]", str(v))
885 steps: list[tuple[str, ...]] = []
886 cur_query: list[str] = []
887 current_endpoint = default_endpoint
888 current_engine = "sparql"
890 directive_handlers = {
891 "with": self._handle_directive_with,
892 "join": self._handle_directive_join,
893 "remove": lambda parts: (None, None, ("REMOVE", parts[1:])),
894 "values": self._handle_directive_values,
895 "page": self._handle_directive_page,
896 }
898 def flush_query() -> None:
899 if cur_query:
900 q = "\n".join(cur_query).strip()
901 if not q:
902 cur_query.clear()
903 return
904 for p, v in params.items():
905 q = q.replace(f"[[{p}]]", str(v))
906 steps.append(("QUERY", current_endpoint, current_engine, q))
907 cur_query.clear()
909 for raw in text.splitlines():
910 line = raw.rstrip("\n")
911 if not self._is_directive(line):
912 cur_query.append(line)
913 continue
915 flush_query()
917 new_endpoint, new_engine, step = self._process_directive(line, directive_handlers)
918 current_endpoint, current_engine = Operation._update_directive_state(
919 current_endpoint,
920 current_engine,
921 new_endpoint,
922 new_engine,
923 )
924 if step is not None:
925 steps.append(step)
927 flush_query()
928 return steps
930 def _send_sparql_csv_request(self, endpoint_url: str, query_text: str) -> Response:
931 headers = {
932 "Accept": "text/csv",
933 "User-Agent": "RAMOSE/2.0.0",
934 }
935 auth = backend_auth(endpoint_url)
936 if self.sparql_http_method == "get":
937 return _http_session.get(
938 endpoint_url + "?query=" + quote(query_text),
939 headers=headers,
940 auth=auth,
941 timeout=self.sparql_timeout,
942 )
943 return _http_session.post(
944 endpoint_url,
945 data=query_text,
946 headers={
947 **headers,
948 "Content-Type": "application/sparql-query",
949 },
950 auth=auth,
951 timeout=self.sparql_timeout,
952 )
954 def _request_sparql_csv(self, endpoint_url: str, query_text: str) -> Response:
955 retry_wait = self.retry_wait
956 for attempt in range(self.retry_attempts):
957 try:
958 response = self._send_sparql_csv_request(endpoint_url, query_text)
959 except (RequestsTimeout, TimeoutError) as exc:
960 if attempt + 1 == self.retry_attempts:
961 msg = f"HTTP status code 408: SPARQL request timeout: {exc}"
962 raise HttpError(HTTPStatus.REQUEST_TIMEOUT, msg) from exc
963 self._sleep_before_retry(retry_wait)
964 retry_wait *= self.retry_backoff
965 continue
966 except RequestException as exc:
967 if attempt + 1 == self.retry_attempts:
968 msg = f"HTTP status code 502: SPARQL request failed: {exc}"
969 raise HttpError(HTTPStatus.BAD_GATEWAY, msg) from exc
970 self._sleep_before_retry(retry_wait)
971 retry_wait *= self.retry_backoff
972 continue
974 response.encoding = "utf-8"
975 if response.status_code not in _RETRYABLE_STATUS_CODES or attempt + 1 == self.retry_attempts:
976 return response
977 self._sleep_before_retry(retry_wait)
978 retry_wait *= self.retry_backoff
980 msg = "SPARQL request did not run"
981 raise RuntimeError(msg)
983 @staticmethod
984 def _sleep_before_retry(retry_wait: float) -> None:
985 if retry_wait:
986 time.sleep(retry_wait)
988 def _run_sparql_dicts(self, endpoint_url: str, query_text: str) -> list[dict[str, object]]:
989 r = self._request_sparql_csv(endpoint_url, query_text)
990 r.encoding = "utf-8"
991 if r.status_code != HTTPStatus.OK:
992 msg = f"SPARQL {r.status_code}: {r.reason}"
993 raise RuntimeError(msg)
994 text = r.content.decode("utf-8-sig", errors="replace")
995 list_of_lines = text.splitlines()
996 return list(DictReader(list_of_lines)) # type: ignore[return-value]
998 @staticmethod
999 def _normalize_sparql_json_resultset(result: dict[str, object]) -> list[dict[str, object]]:
1000 """Convert a SPARQL JSON ResultSet dict to a list of flat dicts."""
1001 vars_ = result["head"].get("vars") or [] # type: ignore[union-attr]
1002 return [
1003 {v: (b[v].get("value") if isinstance(b.get(v), dict) else b.get(v)) for v in vars_}
1004 for b in result["results"].get("bindings", []) # type: ignore[union-attr]
1005 ]
1007 @staticmethod
1008 def _normalize_columnar_dict(result: dict[str, object]) -> list[dict[str, object]]:
1009 """Convert a column-oriented dict {col: [values]} to a list of row dicts."""
1010 cols = list(result.keys())
1011 max_len = max((len(v) for v in result.values() if isinstance(v, (list, tuple))), default=0)
1013 if not max_len:
1014 return [result]
1016 rows = []
1017 for i in range(max_len):
1018 row = {}
1019 for c in cols:
1020 v = result[c]
1021 row[c] = (
1022 v[i]
1023 if isinstance(v, (list, tuple)) and i < len(v)
1024 else (v if not isinstance(v, (list, tuple)) else None)
1025 )
1026 rows.append(row)
1027 return rows
1029 @staticmethod
1030 def _sparql_anything_error_status(exc: Exception) -> int | None:
1031 message = str(exc)
1032 status_match = search(_SPARQL_ANYTHING_HTTP_STATUS_RE, message)
1033 if status_match is not None:
1034 return int(status_match.group(1))
1036 message = message.lower()
1037 if any(marker in message for marker in _SPARQL_ANYTHING_TIMEOUT_MARKERS):
1038 return int(HTTPStatus.REQUEST_TIMEOUT)
1039 if any(marker in message for marker in _SPARQL_ANYTHING_NETWORK_MARKERS):
1040 return int(HTTPStatus.BAD_GATEWAY)
1041 return None
1043 @staticmethod
1044 def _raise_sparql_anything_error(status_code: int, exc: Exception) -> NoReturn:
1045 msg = f"HTTP status code {status_code}: SPARQL Anything request failed: {exc}"
1046 raise HttpError(status_code, msg) from exc
1048 def _request_sparql_anything_select(self, kwargs: dict[str, object]) -> object:
1049 if self._sa_engine is None:
1050 msg = "SPARQL Anything engine not initialized"
1051 raise RuntimeError(msg)
1052 sa_engine = cast("SparqlAnythingEngine", self._sa_engine)
1054 retry_wait = self.retry_wait
1055 for attempt in range(self.retry_attempts):
1056 try:
1057 return sa_engine.select(output_type=dict, **kwargs)
1058 except Exception as exc: # noqa: PERF203
1059 status_code = self._sparql_anything_error_status(exc)
1060 if status_code is None:
1061 raise
1062 if status_code not in _RETRYABLE_STATUS_CODES or attempt + 1 == self.retry_attempts:
1063 self._raise_sparql_anything_error(status_code, exc)
1064 self._sleep_before_retry(retry_wait)
1065 retry_wait *= self.retry_backoff
1067 msg = "SPARQL Anything request did not run"
1068 raise RuntimeError(msg)
1070 def _run_sparql_anything_dicts(
1071 self, query_text: str, values: dict[str, str] | None = None
1072 ) -> list[dict[str, object]]:
1073 """
1074 Execute a SPARQL Anything SELECT query via PySPARQL-Anything and return
1075 a list of dicts (one per row), in the same shape as _run_sparql_dicts.
1077 query_text: full SPARQL (Anything) query string
1078 (typically containing SERVICE <x-sparql-anything:...>).
1079 values: optional dict of template parameters for the query
1080 (name -> value), passed to SPARQL Anything's `values=`.
1081 """
1082 if self._sa_engine is None:
1083 if SparqlAnything is None:
1084 msg = "pysparql_anything not installed. Install with: pip install ramose[sparql-anything]"
1085 raise ImportError(msg)
1086 self._sa_engine = SparqlAnything()
1088 kwargs: dict[str, object] = {"query": query_text}
1089 if values:
1090 kwargs["values"] = {str(k): str(v) for k, v in values.items()} # type: ignore[assignment]
1092 result = self._request_sparql_anything_select(kwargs)
1094 # Normalize to list[dict]
1095 if isinstance(result, list):
1096 if result and isinstance(result[0], dict):
1097 return result
1098 return [dict(row) for row in result]
1100 if not isinstance(result, dict):
1101 return [{"result": result}]
1103 # Standard SPARQL JSON ResultSet shape
1104 head = result.get("head")
1105 results_obj = result.get("results")
1106 if isinstance(head, dict) and isinstance(results_obj, dict) and "bindings" in results_obj:
1107 return self._normalize_sparql_json_resultset(result)
1109 # Column-oriented dict or single-row fallback
1110 return self._normalize_columnar_dict(result)
1112 def _run_query_dicts(self, endpoint_url: str, engine: str, query_text: str) -> list[dict[str, object]]:
1113 if engine == "sparql-anything":
1114 return self._run_sparql_anything_dicts(query_text)
1115 if engine != "sparql":
1116 msg = f"Unknown query engine {engine!r}"
1117 raise ValueError(msg)
1118 return self._run_sparql_dicts(endpoint_url, query_text)
1120 def _inject_values_clause(self, query_text: str, vars_: list[str], acc_rows: list[dict[str, object]] | None) -> str:
1121 # None means no prior step ran: leave the query unrestricted.
1122 # An empty list means a prior step matched nothing: keep going so the empty
1123 # accumulator injects an empty VALUES block, which correctly yields zero solutions.
1124 if acc_rows is None:
1125 return query_text
1127 # build distinct tuples for requested vars from the accumulator
1128 cols = [v.lstrip("?") for v in vars_]
1129 tuples, seen = [], set()
1130 for row in acc_rows:
1131 tup = tuple(row.get(c, "") for c in cols)
1132 if all(tup) and tup not in seen:
1133 seen.add(tup)
1134 tuples.append(tup)
1136 # format literals vs IRIs
1137 def fmt(x: object) -> str:
1138 s = str(x)
1139 if s.startswith(("http://", "https://")):
1140 return f"<{s}>"
1141 return '"' + s.replace("\\", "\\\\").replace('"', '\\"') + '"'
1143 head = "VALUES (" + " ".join(vars_) + ") {\n"
1144 body = "\n".join(" (" + " ".join(fmt(v) for v in tup) + ")" for tup in tuples)
1145 tail = "\n}\n"
1147 i = query_text.find("{")
1148 if i == -1:
1149 # no WHERE brace: put VALUES at top (legal SPARQL)
1150 return head + body + tail + query_text
1151 j = i + 1
1152 return query_text[:j] + "\n" + head + body + tail + query_text[j:]
1154 @staticmethod
1155 def _drop_columns(rows: list[dict[str, object]], vars_: list[str]) -> list[dict[str, object]]:
1156 if not rows:
1157 return rows
1158 vars_set = {v.lstrip("?") for v in vars_}
1159 return [{k: v for k, v in r.items() if k not in vars_set and ("?" + k) not in vars_set} for r in rows]
1161 def _norm_join_key(self, v: object) -> str | None:
1162 if v is None:
1163 return None
1164 s = str(v).strip()
1165 if not s:
1166 return None
1167 # unify scheme for w3id IRIs (and similar)
1168 if s.startswith("http://"):
1169 s = "https://" + s[len("http://") :]
1170 # drop a single trailing slash for stability
1171 return s.removesuffix("/")
1173 @staticmethod
1174 def _merge_row(
1175 left_row: dict[str, object], right_row: dict[str, object], right_cols: list[str]
1176 ) -> dict[str, object]:
1177 merged = dict(left_row)
1178 for col in right_cols:
1179 right_val = right_row.get(col)
1180 if right_val is None:
1181 continue
1182 if col not in merged or merged[col] in ("", None):
1183 merged[col] = right_val
1184 else:
1185 alt = f"{col}_r"
1186 if alt not in merged or merged[alt] in ("", None):
1187 merged[alt] = right_val
1188 return merged
1190 def _join(
1191 self,
1192 left_rows: list[dict[str, object]] | None,
1193 right_rows: list[dict[str, object]] | None,
1194 lkey: str,
1195 rkey: str,
1196 how: str = "inner",
1197 ) -> list[dict[str, object]]:
1198 """
1199 Merge two row sets on lkey (from left_rows) and rkey (from right_rows).
1200 - lkey/rkey may be passed as '?var' or 'var' -> we normalize to bare names.
1201 - Keys are normalized with _norm_join_key (e.g., http -> https, trim slash).
1202 - When 'left', all left rows are preserved even if no match on the right.
1203 - Right-hand columns are copied into the merged row; collisions are avoided.
1204 """
1205 lcol = lkey.lstrip("?")
1206 rcol = rkey.lstrip("?")
1208 left_rows = left_rows or []
1209 right_rows = right_rows or []
1211 rindex: dict[str, list[dict[str, object]]] = {}
1212 for r in right_rows:
1213 rk = self._norm_join_key(r.get(rcol))
1214 if rk is None:
1215 continue
1216 rindex.setdefault(rk, []).append(r)
1218 right_cols = [c for c in (right_rows[0].keys() if right_rows else []) if c != rcol]
1220 out: list[dict[str, object]] = []
1221 for left_row in left_rows:
1222 lk = self._norm_join_key(left_row.get(lcol))
1223 matches = rindex.get(lk, []) # type: ignore[arg-type]
1224 if matches:
1225 out.extend(self._merge_row(left_row, r, right_cols) for r in matches)
1226 elif how == "left":
1227 out.append(dict(left_row))
1228 return out
1230 def _apply_custom_postprocess_params(self, table: ResultTable, q_string: dict[str, list[str]]) -> ResultTable:
1231 for param_name, param_conf in self.custom_params.items():
1232 if param_conf["phase"] != "postprocess":
1233 continue
1234 if param_name in q_string:
1235 handler = getattr(self.addon, param_conf["handler"])
1236 table = handler(table, q_string[param_name])
1237 return table
1239 @property
1240 def _cache_ttl(self) -> int:
1241 if "cache_duration" in self.i:
1242 return int(self.i["cache_duration"])
1243 return self._default_cache_ttl
1245 def _build_cache_key(self, q_string: dict[str, list[str]]) -> str:
1246 presentation_params = {"format", "json"}
1247 if "@@page" not in self.i["sparql"]:
1248 presentation_params |= {"page", "page_size"}
1249 data_params = sorted((name, values) for name, values in q_string.items() if name not in presentation_params)
1250 if data_params:
1251 query_string = "&".join(f"{name}={value}" for name, values in data_params for value in values)
1252 return f"{self.tp}:{self.op_url}?{query_string}"
1253 return f"{self.tp}:{self.op_url}"
1255 def _extract_pagination_params(self, q_string: dict[str, list[str]]) -> tuple[int, int] | None:
1256 page_size_active = self._is_builtin_param_active("page_size")
1257 page_active = self._is_builtin_param_active("page")
1258 if "page_size" not in q_string or not page_size_active:
1259 if page_size_active and page_active and "page" in q_string:
1260 Operation._raise_unprocessable("page requires page_size")
1261 return None
1262 page_size = Operation._parse_positive_int_param(q_string, "page_size")
1263 page = 1
1264 if "page" in q_string and page_active:
1265 page = Operation._parse_positive_int_param(q_string, "page")
1266 return page, page_size
1268 def _has_custom_converter(self, q_string: dict[str, list[str]]) -> bool:
1269 if "format" in q_string and self._is_builtin_param_active("format"):
1270 for req_format in q_string["format"]:
1271 if req_format in self.format:
1272 return True
1273 elif "default_format" in self.i and self.i["default_format"].strip() in self.format:
1274 return True
1275 return False
1277 def _paginate_and_format(
1278 self,
1279 table: ResultTable,
1280 q_string: dict[str, list[str]],
1281 content_type: str,
1282 ) -> OperationResponse:
1283 has_page_directive = "@@page" in self.i["sparql"]
1284 if self._has_custom_converter(q_string) and not has_page_directive:
1285 self.pagination_info = None
1286 # A @@page step already paginated upstream; do not paginate again here.
1287 elif not has_page_directive:
1288 page_params = self._extract_pagination_params(q_string)
1289 if page_params is not None:
1290 page, page_size = page_params
1291 total_items = len(table) - 1
1292 total_pages = ceil(total_items / page_size)
1293 Operation._validate_page_range(page, total_items, total_pages)
1294 start = (page - 1) * page_size
1295 end = start + page_size
1296 table = [table[0], *table[1 + start : 1 + end]]
1297 self.pagination_info = build_pagination_info(self.op_url, q_string, page, page_size, total_items)
1298 else:
1299 self.pagination_info = None
1301 s_res = StringIO()
1302 writer(s_res).writerows(table)
1303 formatted, ctype = self._format_response(s_res.getvalue(), q_string, content_type)
1304 return OperationResponse(formatted.status_code, formatted.body, ctype, {})
1306 def _cache_value(self, rows: ResultTable) -> CachedResult:
1307 pagination: CachedPagination | None = None
1308 if "@@page" in self.i["sparql"] and self.pagination_info is not None:
1309 pagination = {
1310 "page": self.pagination_info.page,
1311 "page_size": self.pagination_info.page_size,
1312 "total_items": self.pagination_info.total_items,
1313 }
1314 return {"rows": rows, "pagination": pagination}
1316 def _format_cached_result(
1317 self,
1318 cached_value: object,
1319 q_string: dict[str, list[str]],
1320 content_type: str,
1321 ) -> OperationResponse:
1322 entry = cast("CachedResult", cached_value)
1323 if entry["pagination"] is not None:
1324 pagination = entry["pagination"]
1325 self.pagination_info = build_pagination_info(
1326 self.op_url,
1327 q_string,
1328 pagination["page"],
1329 pagination["page_size"],
1330 pagination["total_items"],
1331 )
1332 return self._paginate_and_format(entry["rows"], q_string, content_type)
1334 def _finalize_result(
1335 self,
1336 csv_rows: list[list[str]] | list[list[str | object]],
1337 content_type: str,
1338 output_fields: list[str] | None = None,
1339 ) -> OperationResponse:
1340 """Run the shared pipeline: type fields, postprocess, select the output fields, filter, remove types,
1341 cache, paginate, format."""
1342 q_string = parse_qs(quote(self.url_parsed.query, safe="&="))
1343 res = self.type_fields(csv_rows, self.i) # type: ignore[arg-type]
1344 if self.addon is not None:
1345 res = self.postprocess(res, self.i, self.addon)
1346 if output_fields is not None:
1347 res = Operation._select_columns(res, output_fields)
1348 res = self.handling_params(q_string, res)
1349 res = self.remove_types(res)
1350 if self.custom_params:
1351 res = self._apply_custom_postprocess_params(res, q_string)
1352 if self._cache is not None and "cache_disable" not in self.i:
1353 self._cache.set(self._build_cache_key(q_string), self._cache_value(res), expire=self._cache_ttl)
1354 return self._paginate_and_format(res, q_string, content_type)
1356 @staticmethod
1357 def _header_from_field_type(op_item: dict[str, str], acc: list[dict[str, object]]) -> list[str]:
1358 # Respect #field_type order if provided, else derive from data
1359 if "field_type" in op_item:
1360 # FIELD_TYPE_RE is global in this file
1361 return [f for (_, f) in findall(FIELD_TYPE_RE, op_item["field_type"])]
1362 # fallback to keys of first row
1363 return list(acc[0].keys()) if acc else []
1365 @staticmethod
1366 def _select_columns(
1367 res: list[list[str] | list[tuple[object, str]]], fields: list[str]
1368 ) -> list[list[str] | list[tuple[object, str]]]:
1369 header = cast("list[str]", res[0])
1370 indexes = [header.index(field) for field in fields]
1371 return [fields, *([row[idx] for idx in indexes] for row in res[1:])] # type: ignore[list-item]
1373 @staticmethod
1374 def _to_csv_rows(header: list[str], acc: list[dict[str, object]]) -> list[list[object]]:
1375 rows: list[list[object]] = [header] # type: ignore[list-item]
1376 rows.extend([d.get(h, "") for h in header] for d in acc)
1377 return rows
1379 def _extract_params(self, body_params: Mapping[str, object] | None = None) -> dict[str, object]:
1380 """Extract URL parameters (and request body parameters, for write operations) and apply type
1381 conversions based on the operation spec."""
1382 par_dict: dict[str, object] = {}
1383 url_match = match(self.op, self.op_url)
1384 if url_match is None:
1385 msg = f"URL {self.op_url} does not match pattern {self.op}"
1386 raise ValueError(msg)
1387 par_man = url_match.groups()
1388 for idx, par in enumerate(findall("{([^{}]+)}", self.i["url"])):
1389 par_dict[par] = self._bind_value(par, par_man[idx])
1390 for par, val in (body_params or {}).items():
1391 par_dict[par] = self._bind_value(par, val)
1392 return par_dict
1394 def _bind_value(self, param: str, value: object) -> object:
1395 kind = self.i[param].split("(")[0] if param in self.i else "str"
1396 text = "" if value is None else str(value)
1397 if kind == "iri":
1398 return escape_iri(text)
1399 if kind in ("str", "literal"):
1400 return escape_literal(text)
1401 return self.dt.get_func(kind)(text)
1403 @staticmethod
1404 def _apply_config_filters(config: FiltersConfig, values: list[str]) -> dict[str, str]:
1405 try:
1406 return apply_filters(config, values)
1407 except InvalidParameterValueError as exc:
1408 raise HttpError(HTTPStatus.BAD_REQUEST, f"HTTP status code 400: {exc}") from exc
1409 except ValueError as exc:
1410 Operation._raise_unprocessable(str(exc))
1412 def _resolve_preprocess_handler(self, param_name: str, handler: str) -> Callable[[list[str]], dict[str, str]]:
1413 if param_name in self.custom_param_configs:
1414 config = self.custom_param_configs[param_name]
1415 return lambda values: Operation._apply_config_filters(config, values)
1416 return getattr(self.addon, handler)
1418 def _apply_custom_preprocess_params(self, par_dict: dict[str, object]) -> None:
1419 q_string = parse_qs(quote(self.url_parsed.query, safe="&="))
1420 for param_name, param_conf in self.custom_params.items():
1421 if param_conf["phase"] != "preprocess":
1422 continue
1423 if param_name in q_string:
1424 handler = self._resolve_preprocess_handler(param_name, param_conf["handler"])
1425 par_dict.update(handler(q_string[param_name]))
1426 elif param_name not in par_dict:
1427 par_dict[param_name] = ""
1428 for placeholder in findall(r"\[\[(\w+)\]\]", self.i["sparql"]):
1429 if placeholder not in par_dict:
1430 par_dict[placeholder] = ""
1432 def _exec_standard_sparql(self, par_dict: dict[str, object], content_type: str) -> OperationResponse:
1433 """Execute standard SPARQL queries, handling parameter combinations via cartesian product."""
1434 # Wrap scalar values in lists for cartesian product
1435 par_dict = {k: v if isinstance(v, list) else [v] for k, v in par_dict.items()}
1437 parameters_comb = [
1438 dict(zip(par_dict.keys(), combination, strict=False))
1439 for combination in product(*par_dict.values()) # type: ignore[arg-type]
1440 ]
1442 # Example: {"id":"5","area":["A1","A2"]} -> [{"id":"5","area":"A1"}, {"id":"5","area":"A2"}]
1444 list_of_res = []
1445 include_header_line = True
1446 for comb in parameters_comb:
1447 query = self.i["sparql"]
1448 for param, val in comb.items():
1449 query = query.replace(f"[[{param}]]", str(val))
1451 r = self._request_sparql_csv(self.tp, query)
1453 if r.status_code != HTTPStatus.OK:
1454 return OperationResponse.error(r.status_code, f"HTTP status code {r.status_code}: {r.reason}")
1456 # Re-encode to handle non-UTF8 characters in splitlines
1457 list_of_lines = [line.decode("utf-8") for line in r.text.encode("utf-8").splitlines()]
1459 # Include the CSV header only from the first response
1460 if not include_header_line:
1461 list_of_lines = list_of_lines[1:]
1462 include_header_line = False
1464 list_of_res += list_of_lines
1466 return self._finalize_result(list(reader(list_of_res)), content_type)
1468 def _exec_multi_source_query_step(
1469 self, endpoint_url: str, engine: str, qtxt: str, state: dict[str, object]
1470 ) -> None:
1471 """Handle a QUERY step in the multi-source pipeline."""
1472 pending_values = state["pending_values"]
1473 if pending_values:
1474 variables, batch_size, workers = pending_values # type: ignore[misc]
1475 value_batches = self._values_batches(variables, state["acc"], batch_size) # type: ignore[arg-type]
1476 batch_queries = [self._inject_values_clause(qtxt, variables, value_batch) for value_batch in value_batches]
1477 rows = []
1478 with ThreadPoolExecutor(max_workers=workers) as executor:
1479 for batch_rows in executor.map(
1480 lambda batch_query: self._run_query_dicts(endpoint_url, engine, batch_query), batch_queries
1481 ):
1482 rows.extend(batch_rows)
1483 state["pending_values"] = None
1484 else:
1485 rows = self._run_query_dicts(endpoint_url, engine, qtxt)
1487 if state["acc"] is None:
1488 state["acc"] = rows
1489 elif state["pending_join"]:
1490 lvar, rvar, how = state["pending_join"] # type: ignore[misc]
1491 state["acc"] = self._join(state["acc"], rows, lvar, rvar, how) # type: ignore[arg-type]
1492 state["pending_join"] = None
1493 else:
1494 msg = "Multiple QUERY steps without an explicit @@join directive"
1495 raise ValueError(msg)
1497 @staticmethod
1498 def _values_batches(
1499 variables: list[str], acc: list[dict[str, object]] | None, batch_size: int | None
1500 ) -> list[list[dict[str, object]] | None]:
1501 if acc is None:
1502 return [None]
1503 columns = [variable.lstrip("?") for variable in variables]
1504 distinct_rows: list[dict[str, object]] = []
1505 seen: set[tuple[object, ...]] = set()
1506 for row in acc:
1507 if not all(column in row for column in columns):
1508 continue
1509 values = tuple(row[column] for column in columns)
1510 if all(values) and values not in seen:
1511 seen.add(values)
1512 distinct_rows.append(row)
1513 if batch_size is None:
1514 return [distinct_rows]
1515 return [distinct_rows[start : start + batch_size] for start in range(0, len(distinct_rows), batch_size)] or [[]]
1517 def _exec_page_step(self, var: str, default_size: str, max_size: str, state: dict[str, object]) -> None:
1518 q_string = parse_qs(quote(self.url_parsed.query, safe="&="))
1520 page_size_active = self._is_builtin_param_active("page_size")
1521 page_active = self._is_builtin_param_active("page")
1522 explicit_page_size = page_size_active and "page_size" in q_string
1524 if explicit_page_size:
1525 page_size = Operation._parse_positive_int_param(q_string, "page_size")
1526 elif default_size:
1527 page_size = int(default_size)
1528 else:
1529 if page_size_active and page_active and "page" in q_string:
1530 Operation._raise_unprocessable("page requires page_size")
1531 return
1532 if page_size < 1:
1533 msg = f"page_size must be >= 1, got {page_size}"
1534 raise ValueError(msg)
1535 if max_size:
1536 max_page_size = int(max_size)
1537 if explicit_page_size and page_size > max_page_size:
1538 Operation._raise_unprocessable(f"page_size must be <= {max_page_size}, got {page_size}")
1539 page_size = min(page_size, max_page_size)
1541 page = 1
1542 if "page" in q_string and page_active:
1543 page = Operation._parse_positive_int_param(q_string, "page")
1545 column = var.lstrip("?")
1546 acc = state["acc"]
1547 rows = [] if acc is None else cast("list[dict[str, object]]", acc)
1548 distinct = list(dict.fromkeys(row[column] for row in rows if row.get(column)))
1549 total_items = len(distinct)
1550 total_pages = ceil(total_items / page_size)
1551 Operation._validate_page_range(page, total_items, total_pages)
1553 start = (page - 1) * page_size
1554 keep = set(distinct[start : start + page_size])
1555 state["acc"] = [row for row in rows if row.get(column) in keep]
1556 self.pagination_info = build_pagination_info(self.op_url, q_string, page, page_size, total_items)
1558 def _exec_multi_source(self, par_dict: dict[str, object], content_type: str) -> OperationResponse:
1559 """Execute a multi-source query pipeline with @@ directives."""
1560 steps = self._parse_steps(self.i["sparql"], self.tp, par_dict)
1562 state: dict[str, object] = {
1563 "acc": None,
1564 "pending_join": None,
1565 "pending_values": None,
1566 }
1568 for st in steps:
1569 tag = st[0]
1571 if tag == "QUERY":
1572 self._exec_multi_source_query_step(st[1], st[2], st[3], state)
1573 elif tag == "JOIN":
1574 state["pending_join"] = (st[1], st[2], st[3])
1575 elif tag == "REMOVE":
1576 state["acc"] = self._drop_columns(state["acc"] or [], st[1]) # type: ignore[arg-type]
1577 elif tag == "VALUES_INJECT":
1578 state["pending_values"] = (st[1], st[2], st[3])
1579 elif tag == "PAGE":
1580 self._exec_page_step(st[1], st[2], st[3], state)
1581 else:
1582 msg = f"Unknown step tag {tag}"
1583 raise RuntimeError(msg)
1585 acc = cast("list[dict[str, object]]", state["acc"] or [])
1586 output_fields = self._header_from_field_type(self.i, acc)
1587 # Postprocess functions also see the step columns missing from #field_type;
1588 # the output keeps only the declared ones.
1589 step_fields = [key for key in dict.fromkeys(key for row in acc for key in row) if key not in output_fields]
1590 csv_rows = self._to_csv_rows(output_fields + step_fields, acc)
1591 return self._finalize_result(csv_rows, content_type, output_fields)
1593 @staticmethod
1594 def _format_error(sc: int, e: Exception, prefix: str = "") -> OperationResponse:
1595 """Format an error response with traceback line info."""
1596 tb = e.__traceback__
1597 line = tb.tb_lineno if tb else "?"
1598 msg = f"HTTP status code {sc}: {prefix}{type(e).__name__}: {e} (line {line})"
1599 return OperationResponse.error(sc, msg)
1601 def exec(
1602 self,
1603 method: str = "get",
1604 content_type: str = "application/json",
1605 body_params: Mapping[str, object] | None = None,
1606 ) -> OperationResponse:
1607 """This method takes in input the HTTP method to use for the call
1608 and the content type to return, and execute the operation as indicated
1609 in the specification file, by running (in the following order):
1611 1. the methods to preprocess the query;
1612 2. the SPARQL query related to the operation called, by using the parameters indicated in the URL;
1613 3. the specification of all the types of the various rows returned;
1614 4. the methods to postprocess the result;
1615 5. the application of the filter to remove, filter, sort the result;
1616 6. the removal of the types added at the step 3, so as to have a data structure ready to be returned;
1617 7. the conversion in the format requested by the user."""
1618 str_method = method.lower()
1619 if str_method not in self.i["method"].split():
1620 return OperationResponse.error(405, f"HTTP status code 405: '{str_method}' method not allowed")
1622 try:
1623 if self._is_write(str_method):
1624 response = self._exec_update(self._prepare_params(body_params), content_type)
1625 else:
1626 response = self._dispatch_exec(content_type, body_params)
1627 except HttpError as err:
1628 return OperationResponse.error(err.status_code, str(err))
1629 except TimeoutError as e:
1630 return self._format_error(408, e, "request timeout - ")
1631 except (TypeError, ValueError) as e:
1632 return self._format_error(400, e, "parameter in the request not compliant with the type specified - ")
1633 except Exception as e: # noqa: BLE001
1634 return self._format_error(500, e, "something unexpected happened - ")
1636 headers = dict(response.headers)
1637 if self.pagination_info is not None:
1638 link_header = build_link_header(self.pagination_info)
1639 if link_header:
1640 headers["Link"] = link_header
1641 return OperationResponse(
1642 response.status_code,
1643 response.body,
1644 response.content_type,
1645 headers,
1646 response.is_error_message,
1647 )
1649 def _prepare_params(self, body_params: Mapping[str, object] | None = None) -> dict[str, object]:
1650 par_dict = self._extract_params(body_params)
1651 if self.addon is not None:
1652 self.preprocess(par_dict, self.i, self.addon)
1653 if self.custom_params:
1654 self._apply_custom_preprocess_params(par_dict)
1655 return par_dict
1657 def _dispatch_exec(
1658 self,
1659 content_type: str,
1660 body_params: Mapping[str, object] | None = None,
1661 ) -> OperationResponse:
1662 """Dispatch to the appropriate read execution path based on the SPARQL text content."""
1663 par_dict = self._prepare_params(body_params)
1665 if self._cache is not None and "cache_disable" not in self.i:
1666 q_string = parse_qs(quote(self.url_parsed.query, safe="&="))
1667 cached_table = self._cache.get(self._build_cache_key(q_string))
1668 if cached_table is not None:
1669 return self._format_cached_result(cached_table, q_string, content_type)
1671 sparql_text = self.i["sparql"]
1672 resolved_text = sparql_text
1673 for param, val in par_dict.items():
1674 resolved_text = resolved_text.replace(f"[[{param}]]", str(val))
1676 if "@@" not in resolved_text:
1677 return self._exec_standard_sparql(par_dict, content_type)
1679 try:
1680 return self._exec_multi_source(par_dict, content_type)
1681 except ValueError as ve:
1682 return OperationResponse.error(400, f"HTTP status code 400: {ve}")
1683 except RuntimeError as re_err:
1684 return OperationResponse.error(502, f"HTTP status code 502: {re_err}")
1686 @staticmethod
1687 def _is_write(method: str) -> bool:
1688 return method.lower() in _WRITE_METHODS
1690 def _format_write_success(self, content_type: str) -> OperationResponse:
1691 if content_type == "text/csv":
1692 return OperationResponse(HTTPStatus.OK, "status,message\r\n200,operation completed\r\n", "text/csv", {})
1693 return OperationResponse(
1694 HTTPStatus.OK,
1695 dumps({"status": 200, "message": "operation completed"}),
1696 "application/json",
1697 {},
1698 )
1700 def _exec_update(self, par_dict: dict[str, object], content_type: str) -> OperationResponse:
1701 """Send a SPARQL 1.1 Update to the update endpoint and return a confirmation with no result set."""
1702 update_text = self.i["sparql"]
1703 for param, val in par_dict.items():
1704 update_text = update_text.replace(f"[[{param}]]", str(val))
1706 unresolved = findall(r"\[\[(\w+)\]\]", update_text)
1707 if unresolved:
1708 missing = ", ".join(dict.fromkeys(unresolved))
1709 message = f"HTTP status code 400: missing required parameter(s): {missing}"
1710 return OperationResponse.error(HTTPStatus.BAD_REQUEST, message)
1712 endpoint = self.update_endpoint or self.tp
1713 try:
1714 response = _http_session.post(
1715 endpoint,
1716 data={"update": update_text},
1717 headers={"Accept": "application/json"},
1718 auth=backend_auth(endpoint),
1719 timeout=self.sparql_timeout,
1720 )
1721 except RequestException as exc:
1722 msg = f"SPARQL update request failed: {exc}"
1723 raise RuntimeError(msg) from exc
1725 if response.status_code not in (HTTPStatus.OK, HTTPStatus.CREATED, HTTPStatus.NO_CONTENT):
1726 return OperationResponse.error(
1727 response.status_code, f"HTTP status code {response.status_code}: {response.reason}"
1728 )
1730 if self._cache is not None:
1731 self._cache.clear()
1732 return self._format_write_success(content_type)