Coverage for src/time_agnostic_library/qlever_setup.py: 31%

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

2# 

3# SPDX-License-Identifier: ISC 

4 

5import argparse 

6import logging 

7import shutil 

8import subprocess 

9from collections.abc import Iterator 

10from pathlib import Path 

11from typing import cast 

12 

13from pyoxigraph import Literal, NamedNode, Quad, RdfFormat, parse, serialize 

14from sparqlite import SPARQLClient 

15 

16from time_agnostic_library.prov_entity import ProvEntity 

17from time_agnostic_library.qlever import generate_qlever_index 

18from time_agnostic_library.support import generate_config_file 

19 

20logger = logging.getLogger(__name__) 

21 

22 

23def _iter_qlever_input( 

24 dataset_files: list[Path], provenance_files: list[Path] 

25) -> Iterator[Quad]: 

26 predicate = NamedNode(ProvEntity.iri_has_update_query) 

27 graph = NamedNode("urn:tal:qlever:prov/") 

28 for path in provenance_files: 

29 for quad in parse(path=path, format=RdfFormat.N_QUADS): 

30 yield quad 

31 if quad.predicate == predicate: 

32 associations = generate_qlever_index( 

33 [ 

34 ( 

35 cast("NamedNode", quad.subject).value, 

36 cast("Literal", quad.object).value, 

37 ) 

38 ] 

39 ) 

40 for s, p, o in associations: 

41 yield Quad( 

42 NamedNode(str(s)), NamedNode(str(p)), Literal(str(o)), graph 

43 ) 

44 for path in dataset_files: 

45 yield from parse(path=path, format=RdfFormat.N_QUADS) 

46 

47 

48def setup_qlever( 

49 dataset_files: list[Path], 

50 provenance_files: list[Path], 

51 output_dir: Path, 

52 port: int = 7000, 

53) -> Path: 

54 qlever = shutil.which("qlever") 

55 if qlever is None: 

56 msg = 'Install QLever support with uv add "time-agnostic-library[qlever]"' 

57 raise FileNotFoundError(msg) 

58 for path in [*dataset_files, *provenance_files]: 

59 if not path.is_file(): 

60 raise FileNotFoundError(path) 

61 output_dir = output_dir.resolve() 

62 output_dir.mkdir(parents=True) 

63 logger.info("Preparing dataset, provenance and snapshot URI associations") 

64 serialize( 

65 _iter_qlever_input(dataset_files, provenance_files), 

66 output=output_dir / "tal.nq", 

67 format=RdfFormat.N_QUADS, 

68 ) 

69 logger.info("Building QLever index in %s", output_dir) 

70 subprocess.run( # noqa: S603 

71 [ 

72 qlever, 

73 "index", 

74 "--name", 

75 "tal", 

76 "--format", 

77 "nq", 

78 "--input-files", 

79 "tal.nq", 

80 "--cat-input-files", 

81 "cat tal.nq", 

82 "--settings-json", 

83 '{"prefixes-external": []}', 

84 "--system", 

85 "docker", 

86 ], 

87 cwd=output_dir, 

88 check=True, 

89 ) 

90 logger.info("Starting QLever on port %s", port) 

91 subprocess.run( # noqa: S603 

92 [ 

93 qlever, 

94 "start", 

95 "--name", 

96 "tal", 

97 "--description", 

98 "TAL dataset", 

99 "--host-name", 

100 "localhost", 

101 "--port", 

102 str(port), 

103 "--server-container", 

104 f"tal-qlever-{port}", 

105 "--access-token", 

106 "", 

107 "--system", 

108 "docker", 

109 "--no-warmup", 

110 ], 

111 cwd=output_dir, 

112 check=True, 

113 ) 

114 endpoint = f"http://localhost:{port}" 

115 with SPARQLClient(endpoint) as client: 

116 client.ask("ASK { ?s ?p ?o }") 

117 config_path = output_dir / "config.json" 

118 generate_config_file( 

119 str(config_path), 

120 dataset_urls=[endpoint], 

121 provenance_urls=[endpoint], 

122 qlever_full_text_search=True, 

123 ) 

124 logger.info("Endpoint: %s; TAL configuration: %s", endpoint, config_path) 

125 return config_path 

126 

127 

128def main(): # pragma: no cover 

129 parser = argparse.ArgumentParser( 

130 description="Build and serve a QLever index for Time Agnostic Library." 

131 ) 

132 parser.add_argument( 

133 "--dataset", 

134 nargs="+", 

135 type=Path, 

136 required=True, 

137 help="Current dataset files in N-Quads format", 

138 ) 

139 parser.add_argument( 

140 "--provenance", 

141 nargs="+", 

142 type=Path, 

143 required=True, 

144 help="OCDM provenance files in N-Quads format", 

145 ) 

146 parser.add_argument( 

147 "--output-dir", 

148 type=Path, 

149 required=True, 

150 help="New directory for the index and TAL configuration", 

151 ) 

152 parser.add_argument("--port", type=int, default=7000) 

153 args = parser.parse_args() 

154 logging.basicConfig(level=logging.INFO, format="%(message)s") 

155 setup_qlever(args.dataset, args.provenance, args.output_dir, args.port) 

156 

157 

158if __name__ == "__main__": # pragma: no cover 

159 main()