Coverage for oc_ocdm / abstract_set.py: 96%

1#!/usr/bin/python 

2 

3# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com> 

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

5# 

6# SPDX-License-Identifier: ISC 

7 

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

9from __future__ import annotations 

10 

11from abc import ABC, abstractmethod 

12from typing import TYPE_CHECKING, Generic, TypeVar 

13 

14from oc_ocdm.abstract_entity import AbstractEntity 

15 

16if TYPE_CHECKING: 

17 from typing import List, ClassVar, Dict, Optional 

18 from rdflib import URIRef, Graph 

19 

20E = TypeVar('E', bound=AbstractEntity) 

21 

22 

23class AbstractSet(ABC, Generic[E]): 

24 """ 

25 Abstract class which represents a generic set of entities. 

26 It is the base class for each concrete set of entities. 

27 """ 

28 

29 short_name_to_type_iri: ClassVar[Dict[str, URIRef]] = {} 

30 

31 def __init__(self) -> None: 

32 """ 

33 Constructor of the ``AbstractSet`` class. 

34 """ 

35 self.res_to_entity: Dict[URIRef, E] = {} 

36 

37 def graphs(self) -> List[Graph]: 

38 """ 

39 A utility method that allows to retrieve the list of ``rdflib.Graph`` 

40 instances corresponding to each entity contained in the set. 

41 

42 :return: The requested list of graphs 

43 """ 

44 result: List[Graph] = [] 

45 for entity in self.res_to_entity.values(): 

46 if len(entity.g) > 0: 

47 result.append(entity.g) 

48 return result 

49 

50 def __getstate__(self) -> dict[str, object]: 

51 """ 

52 Support for pickle serialization. 

53 

54 RDFLib Graph objects contain threading locks that are not directly picklable, 

55 but Python's pickle module can handle them. This method provides standard 

56 pickle protocol support for AbstractSet instances. 

57 """ 

58 return vars(self).copy() 

59 

60 def __setstate__(self, state: dict[str, object]) -> None: 

61 """ 

62 Support for pickle deserialization. 

63 

64 Restores the AbstractSet state from a pickled representation. 

65 """ 

66 vars(self).update(state) 

67 

68 @abstractmethod 

69 def get_entity(self, res: URIRef) -> Optional[E]: 

70 """ 

71 Method signature for concrete implementations that allow 

72 to retrieve a contained entity identified by its URI. 

73 

74 :param res: The URI that identifies the requested entity 

75 :type res: URIRef 

76 :return: The requested entity if found, None otherwise 

77 """ 

78 raise NotImplementedError 

79 

80 @staticmethod 

81 def get_graph_iri(g: Graph) -> str: 

82 """ 

83 A utility method that allows to retrieve the IRI which represents 

84 the name of a given named graph. 

85 

86 **NOTE: this is a static function!** 

87 

88 :param g: The named graph whose name will be returned 

89 :type g: Graph 

90 :return: The requested string whose content is the IRI associated to the given named graph 

91 """ 

92 return str(g.identifier)