Coverage for oc_ocdm/support/query_utils.py: 80%

1#!/usr/bin/python 

2# -*- coding: utf-8 -*- 

3# Copyright (c) 2016, Silvio Peroni <essepuntato@gmail.com> 

4# 

5# Permission to use, copy, modify, and/or distribute this software for any purpose 

6# with or without fee is hereby granted, provided that the above copyright notice 

7# and this permission notice appear in all copies. 

8# 

9# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH 

10# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND 

11# FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, 

12# OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, 

13# DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS 

14# ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS 

15# SOFTWARE. 

16from __future__ import annotations 

17 

18from typing import TYPE_CHECKING 

19 

20if TYPE_CHECKING: 

21 from typing import Tuple 

22 from rdflib import URIRef 

23 from rdflib.compare import IsomorphicGraph 

24 from oc_ocdm.abstract_entity import AbstractEntity 

25 

26from rdflib import Graph 

27from rdflib.compare import graph_diff, to_isomorphic 

28 

29 

30def get_delete_query(graph_iri: URIRef, data: Graph) -> Tuple[str, int]: 

31 num_of_statements: int = len(data) 

32 if num_of_statements <= 0: 

33 return "", 0 

34 else: 

35 statements: str = data.serialize(format="nt11").replace('\n', '') 

36 delete_string: str = f"DELETE DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

37 return delete_string, num_of_statements 

38 

39 

40def get_insert_query(graph_iri: URIRef, data: Graph) -> Tuple[str, int]: 

41 num_of_statements: int = len(data) 

42 if num_of_statements <= 0: 

43 return "", 0 

44 else: 

45 statements: str = data.serialize(format="nt11").replace('\n', '') 

46 insert_string: str = f"INSERT DATA {{ GRAPH <{graph_iri}> {{ {statements} }} }}" 

47 return insert_string, num_of_statements 

48 

49 

50def get_update_query(entity: AbstractEntity, entity_type: str = "graph") -> Tuple[str, int, int]: 

51 if entity_type in ["graph", "metadata"]: 

52 to_be_deleted: bool = entity.to_be_deleted 

53 preexisting_graph: Graph = entity.preexisting_graph 

54 elif entity_type == "prov": 

55 to_be_deleted: bool = False 

56 preexisting_graph: Graph = Graph(identifier=entity.g.identifier) 

57 

58 if to_be_deleted: 

59 delete_string, removed_triples = get_delete_query(entity.g.identifier, preexisting_graph) 

60 if delete_string != "": 

61 return delete_string, 0, removed_triples 

62 else: 

63 return "", 0, 0 

64 else: 

65 preexisting_iso: IsomorphicGraph = to_isomorphic(preexisting_graph) 

66 current_iso: IsomorphicGraph = to_isomorphic(entity.g) 

67 if preexisting_iso == current_iso: 

68 # Both graphs have exactly the same content! 

69 return "", 0, 0 

70 in_both, in_first, in_second = graph_diff(preexisting_iso, current_iso) 

71 delete_string, removed_triples = get_delete_query(entity.g.identifier, in_first) 

72 insert_string, added_triples = get_insert_query(entity.g.identifier, in_second) 

73 if delete_string != "" and insert_string != "": 

74 return delete_string + '; ' + insert_string, added_triples, removed_triples 

75 elif delete_string != "": 

76 return delete_string, 0, removed_triples 

77 elif insert_string != "": 

78 return insert_string, added_triples, 0 

79 else: 

80 return "", 0, 0