Coverage for src/time_agnostic_library/qlever.py: 100%

1# SPDX-FileCopyrightText: 2026 Arcangelo Massari <arcangelo.massari@unibo.it> 

2# 

3# SPDX-License-Identifier: ISC 

4 

5from collections.abc import Iterable 

6 

7from rdflib import Literal, URIRef 

8 

9from time_agnostic_library.agnostic_entity import _fast_parse_update 

10 

11QLEVER_HAS_WORD = URIRef("http://qlever.cs.uni-freiburg.de/builtin-functions/has-word") 

12 

13 

14def qlever_search_tokens(triple: tuple[str, ...]) -> set[str]: 

15 return { 

16 f"{position}|{term[1:-1]}" 

17 for position, term in zip( 

18 ("subject", "predicate", "object"), triple[:3], strict=True 

19 ) 

20 if term.startswith("<") and term.endswith(">") 

21 } 

22 

23 

24def generate_qlever_index( 

25 snapshot_updates: Iterable[tuple[str, str]], 

26) -> set[tuple[URIRef, URIRef, Literal]]: 

27 return { 

28 (URIRef(snapshot), QLEVER_HAS_WORD, Literal(token)) 

29 for snapshot, update in snapshot_updates 

30 for _, quads in _fast_parse_update(update) 

31 for quad in quads 

32 for token in qlever_search_tokens(quad) 

33 }