Coverage for rdflib_ocdm / support.py: 98%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import re 

10from typing import Match, Optional 

11 

12from rdflib import RDF, XSD, Dataset, Graph, Literal, URIRef 

13 

14prov_regex: str = r"^(.+)/prov/([a-z][a-z])/([1-9][0-9]*)$" 

15 

16 

17def _get_match(regex: str, group: int, string: str) -> str: 

18 match: Match[str] | None = re.match(regex, string) 

19 if match is not None: 

20 return match.group(group) 

21 else: 

22 return "" 

23 

24 

25def is_string_empty(string: str | None) -> bool: 

26 return string is None or string.strip() == "" 

27 

28 

29def get_prov_count(res: URIRef) -> Optional[str]: 

30 string_iri: str = str(res) 

31 if "/prov/" in string_iri: 

32 return _get_match(prov_regex, 3, string_iri) 

33 return None 

34 

35 

36def get_entity_subgraph(graph: Dataset | Graph, entity: URIRef) -> Dataset | Graph: 

37 if isinstance(graph, Dataset): 

38 subj_graph: Dataset = Dataset() 

39 for quad in graph.quads((entity, None, None, None)): 

40 subj_graph.add(quad) # type: ignore[arg-type] 

41 return subj_graph 

42 subj_graph_g: Graph = Graph() 

43 for triple in graph.triples((entity, None, None)): 

44 subj_graph_g.add(triple) 

45 return subj_graph_g 

46 

47 

48def create_literal( 

49 g: Graph, 

50 res: URIRef, 

51 p: URIRef, 

52 s: str | None, 

53 dt: URIRef | None = None, 

54 nor: bool = True, 

55) -> None: 

56 if not is_string_empty(s): 

57 datatype = dt if dt is not None else XSD.string 

58 g.add((res, p, Literal(s, datatype=datatype, normalize=nor))) 

59 

60 

61def create_type( 

62 g: Dataset | Graph, res: URIRef, res_type: URIRef, identifier: str | None = None 

63) -> None: 

64 if isinstance(g, Dataset): 

65 g.add((res, RDF.type, res_type, identifier)) # type: ignore[arg-type] 

66 elif isinstance(g, Graph): 66 ↛ exitline 66 didn't return from function 'create_type' because the condition on line 66 was always true

67 g.add((res, RDF.type, res_type))