bench_executor.query

  1#!/usr/bin/env python3
  2
  3"""
  4Query executes SPARQL queries on endpoints by posting the SPARQL query over
  5HTTP onto the endpoint. It applies timeouts to these queries automatically and
  6checks if the results are empty or not.
  7"""
  8
  9import os
 10import requests
 11from typing import Optional
 12from timeout_decorator import timeout, TimeoutError  # type: ignore
 13from bench_executor.logger import Logger
 14
 15TIMEOUT = 1 * 3600  # 1 hour
 16
 17
 18class Query():
 19    """Execute a query on a SPARQL endpoint."""
 20    def __init__(self, data_path: str, config_path: str, directory: str,
 21                 verbose: bool):
 22        """Creates an instance of the Query class.
 23
 24        Parameters
 25        ----------
 26        data_path : str
 27            Path to the data directory of the case.
 28        config_path : str
 29            Path to the config directory of the case.
 30        directory : str
 31            Path to the directory to store logs.
 32        verbose : bool
 33            Enable verbose logs.
 34        """
 35        self._data_path = os.path.abspath(data_path)
 36        self._config_path = os.path.abspath(config_path)
 37        self._logger = Logger(__name__, directory, verbose)
 38
 39        os.umask(0)
 40        os.makedirs(os.path.join(self._data_path, 'query'), exist_ok=True)
 41
 42    @property
 43    def name(self):
 44        """Name of the class: Query"""
 45        return __name__
 46
 47    @property
 48    def root_mount_directory(self) -> str:
 49        """Subdirectory in the root directory of the case for Query.
 50
 51        Returns
 52        -------
 53        subdirectory : str
 54            Subdirectory of the root directory for Query.
 55
 56        """
 57        return __name__.lower()
 58
 59    @timeout(TIMEOUT)
 60    def _execute_with_timeout(self, query: str, sparql_endpoint: str,
 61                              headers: dict = {}) -> str:
 62        """Execute a query with a provided timeout.
 63
 64        Parameters
 65        ----------
 66        query : str
 67            The query to execute.
 68        sparql_endpoint : str
 69            The URL of the SPARQL endpoint.
 70        headers : dict
 71            HTTP headers to supply when posting the query.
 72
 73        Returns
 74        -------
 75        success : bool
 76            Whether the execution was successfull or not.
 77        """
 78        self._logger.info(f'Executing query "{query}" on endpoint '
 79                          f'"{sparql_endpoint}"')
 80        data = {
 81            'query': query,
 82            'maxrows': '3000000'  # Overwrite Virtuoso SPARQL limit
 83        }
 84        # Hardcoded to N-Triples
 85        r = requests.post(sparql_endpoint, data=data, headers=headers)
 86        if r.status_code != 200:
 87            msg = f'Query failed: {r.text} (HTTP {r.status_code})'
 88            self._logger.error(msg)
 89        r.raise_for_status()
 90        return r.text
 91
 92    def _execute(self, query: str, sparql_endpoint: str, expect_empty: bool,
 93                 headers: dict = {}) -> Optional[str]:
 94        """Execute a query on a SPARQL endpoint
 95
 96        Parameters
 97        ----------
 98        query : str
 99            The query to execute.
100        sparql_endpoint : str
101            The URL of the SPARQL endpoint.
102        expect_empty : bool
103            Whether the expected results are empty or not.
104        headers : dict
105            HTTP headers to supply when posting the query.
106
107        Returns
108        -------
109        results : str
110            The HTTP response as string of the SPARQL endpoint, unless it has
111            no results.
112        """
113        results = None
114        try:
115            results = self._execute_with_timeout(query,
116                                                 sparql_endpoint,
117                                                 headers)
118        except TimeoutError:
119            msg = f'Timeout ({TIMEOUT}s) reached for Query: "{query}"'
120            self._logger.warning(msg)
121
122        # Check results output
123        if results is None or not results or 'Empty' in results:
124            if expect_empty:
125                self._logger.info('No results found, but was expected!')
126                return None
127
128            self._logger.error('No results found!')
129            return None
130
131        return results
132
133    def execute_and_save(self, query: str, sparql_endpoint: str,
134                         results_file: str, expect_empty: bool = False,
135                         headers: dict = {}) -> bool:
136        """Executes a SPARQL query and save the results.
137
138        The results are saved to the `results_file` path.
139
140        Parameters
141        ----------
142        query : str
143            The query to execute.
144        sparql_endpoint : str
145            The URL of the SPARQL endpoint.
146        results_file : str
147            Path to the file where the results may be stored.
148        expect_empty : bool
149            Whether the expected results are empty or not.
150        headers : dict
151            HTTP headers to supply when posting the query.
152
153        Returns
154        -------
155        success : bool
156            Whether the execution succeeded or not.
157        """
158        try:
159            results = self._execute(query, sparql_endpoint, expect_empty,
160                                    headers)
161        except Exception as e:
162            msg = f'Failed to execute query "{query}" on endpoint ' + \
163                  f'"{sparql_endpoint}": {e}'
164            self._logger.error(msg)
165            return False
166
167        path = os.path.join(self._data_path, 'shared')
168        os.umask(0)
169        os.makedirs(path, exist_ok=True)
170
171        if results is not None:
172            results_file = os.path.join(path, results_file)
173            with open(results_file, 'w') as f:
174                f.write(results)
175
176            self._logger.debug(f'Wrote query results to "{results_file}"')
177            self._logger.debug('Query results:')
178            self._logger.debug(results)
179            return True
180
181        return False
182
183    def _read_query_file(self, query_file: str) -> str:
184        """Read the query file
185
186        Returns
187        -------
188        content : str
189            The content of the query file.
190
191        Raises
192        ------
193        FileNotFoundError : Exception
194            If the query file cannot be found.
195        """
196        path = os.path.join(self._data_path, 'shared', query_file)
197        if not os.path.exists(path):
198            msg = f'Query file "{path}" does not exist'
199            self._logger.error(msg)
200            raise FileNotFoundError(msg)
201
202        with open(path, 'r') as f:
203            query = f.read()
204
205        return query
206
207    def execute_from_file(self, query_file: str, sparql_endpoint: str,
208                          expect_empty: bool = False,
209                          headers: dict = {}) -> str:
210        """Executes a SPARQL query from file.
211
212        The results are saved to the `results_file` path.
213
214        Parameters
215        ----------
216        query_file : str
217            Path to the file containing the query.
218        sparql_endpoint : str
219            The URL of the SPARQL endpoint.
220        expect_empty : bool
221            Whether the expected results are empty or not.
222        headers : dict
223            HTTP headers to supply when posting the query.
224
225        Returns
226        -------
227        results : str
228            The HTTP response as string of the SPARQL endpoint, unless it has
229            no results.
230
231        Raises
232        ------
233        Exception : Exception
234            Pass through the exception from the Python's request module
235            regarding HTTP status codes.
236        """
237        query = self._read_query_file(query_file)
238        try:
239            results = self._execute(query, sparql_endpoint, expect_empty,
240                                    headers)
241        except Exception as e:
242            msg = f'Failed to execute query "{query}" on endpoint ' + \
243                  f'"{sparql_endpoint}": {e}'
244            self._logger.error(msg)
245            raise e
246
247        if results is not None:
248            return results
249
250        return ''
251
252    def execute_from_file_and_save(self, query_file: str,
253                                   sparql_endpoint: str,
254                                   results_file: str,
255                                   expect_empty: bool = False,
256                                   headers: dict = {}) -> bool:
257        """Executes a SPARQL query from file and save the results.
258
259        The results are saved to the `results_file` path.
260
261        Parameters
262        ----------
263        query_file : str
264            Path to the file containing the query.
265        sparql_endpoint : str
266            The URL of the SPARQL endpoint.
267        results_file : str
268            Path to the file where the results may be stored.
269        expect_empty : bool
270            Whether the expected results are empty or not.
271        headers : dict
272            HTTP headers to supply when posting the query.
273
274        Returns
275        -------
276        success : bool
277            Whether the execution succeeded or not.
278
279        Raises
280        ------
281        FileNotFoundError : Exception
282            If the query file cannot be found.
283        """
284        query = self._read_query_file(query_file)
285        results = self.execute_and_save(query, sparql_endpoint, results_file,
286                                        expect_empty, headers)
287        if results is not None:
288            return True
289
290        return False