Coverage for lode / helpers / rdf.py: 77%

1"""RDF helpers shared across LODE: IRIs and graphs. 

2 

3Stateless utilities with no knowledge of the LODE pipeline (no models, no reader 

4phases, no network): they take IRIs or rdflib graphs and return plain values. 

5Pipeline types built on top of them (e.g. ProvenanceGraph) live with the reader. 

6""" 

7from pathlib import Path 

8from typing import Optional 

9 

10from rdflib import Graph, URIRef 

11from rdflib.namespace import OWL, RDF, SKOS 

12 

13# ---------------------------------------------------------- 

14# IRIs 

15# ---------------------------------------------------------- 

16 

17 

18def iri_namespace(iri) -> str: 

19 """Namespace of an IRI, trailing separator included. 

20 

21 Splits on the last '#' if the IRI has one, otherwise on the last '/': 

22 'http://ex.org/onto#T' -> 'http://ex.org/onto#' 

23 'http://ex.org/voc/t' -> 'http://ex.org/voc/' 

24 """ 

25 s = str(iri) 

26 if "#" in s: 

27 return s.rsplit("#", 1)[0] + "#" 

28 return s.rsplit("/", 1)[0] + "/" 

29 

30 

31def iri_local_name(iri) -> str: 

32 """Local name of an IRI: the last segment after '#' and '/', trailing '/' ignored. 

33 Never contains '#' or '/', so it is safe as a file name or attribute name: 

34 'http://ex.org/onto#T' -> 'T' 

35 'http://ex.org/voc/t/' -> 't' 

36 """ 

37 return str(iri).rstrip("/").split("#")[-1].split("/")[-1] 

38 

39 

40def iri_stem(iri) -> str: 

41 """IRI without trailing '/' or '#', to compare IRIs that differ only in the 

42 separator ('http://ex.org/onto#' == 'http://ex.org/onto' == 'http://ex.org/onto/').""" 

43 return str(iri).rstrip("/#") 

44 

45 

46def source_iri(source: str) -> str: 

47 """IRI of a load source: URLs as they are, local paths as absolute file: URIs.""" 

48 if "://" in source: 

49 return source 

50 return Path(source).resolve().as_uri() 

51 

52 

53# ---------------------------------------------------------- 

54# Graphs 

55# ---------------------------------------------------------- 

56 

57# Node types that name a document: owl:Ontology first, then skos:ConceptScheme. 

58ONTOLOGY_TYPES = (OWL.Ontology, SKOS.ConceptScheme) 

59 

60 

61def declared_ontology_iri(g: Graph) -> Optional[URIRef]: 

62 """IRI of the ontology (or concept scheme) a document declares, or None. 

63 owl:Ontology wins over skos:ConceptScheme; blank nodes are ignored.""" 

64 for t in ONTOLOGY_TYPES: 

65 for s in g.subjects(RDF.type, t): 

66 if isinstance(s, URIRef): 

67 return s 

68 return None 

69 

70# Formats tried, in order, when the serialization is unknown. Turtle comes before N3 

71# (N3 is a superset and slower to fail); N-Triples is also valid Turtle. 

72PARSE_FORMATS = ("xml", "turtle", "json-ld", "nt", "n3") 

73 

74 

75def parse_any(data, preferred: Optional[str] = None) -> Optional[Graph]: 

76 """Parse RDF `data` (str or bytes) of unknown serialization. 

77 

78 Tries `preferred` first (e.g. the format declared by the Content-Type), then 

79 PARSE_FORMATS in order. Returns the first graph that parses, or None if no 

80 format works: raising a meaningful error is left to the caller.""" 

81 formats = ([preferred] if preferred else []) + [f for f in PARSE_FORMATS if f != preferred] 

82 for fmt in formats: 

83 g = Graph() 

84 try: 

85 g.parse(data=data, format=fmt) 

86 return g 

87 except Exception: 

88 continue 

89 return None