Coverage for rdflib_ocdm / support.py: 96%

34 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-03-21 12:35 +0000

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

7from __future__ import annotations 

8 

9import re 

10from typing import Match 

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 

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

17 match: Match = re.match(regex, string) 

18 if match is not None: 

19 return match.group(group) 

20 else: 

21 return "" 

22 

23def is_string_empty(string: str) -> bool: 

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

25 

26def get_prov_count(res: URIRef) -> str: 

27 string_iri: str = str(res) 

28 if "/prov/" in string_iri: 

29 return _get_match(prov_regex, 3, string_iri) 

30 

31def get_entity_subgraph(graph: Graph, entity: URIRef) -> Graph: 

32 subj_graph: Dataset|Graph = Dataset() if isinstance(graph, Dataset) else Graph() 

33 if isinstance(graph, Dataset): 

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

35 subj_graph.add(quad) 

36 elif isinstance(graph, Graph): 36 ↛ 39line 36 didn't jump to line 39 because the condition on line 36 was always true

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

38 subj_graph.add(triple) 

39 return subj_graph 

40 

41def create_literal(g: Graph, res: URIRef, p: URIRef, s: str, dt: URIRef = None, nor: bool = True) -> None: 

42 if not is_string_empty(s): 

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

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

45 

46def create_type(g: Dataset|Graph, res: URIRef, res_type: URIRef, identifier: str = None) -> None: 

47 if isinstance(g, Dataset): 

48 g.add((res, RDF.type, res_type, identifier)) 

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

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