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 triplelite import TripleLite 

15 

16from oc_ocdm.abstract_entity import AbstractEntity 

17 

18if TYPE_CHECKING: 

19 from typing import ClassVar, Dict, List, Optional 

20 

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

22 

23 

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

25 """ 

26 Abstract class which represents a generic set of entities. 

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

28 """ 

29 

30 short_name_to_type_iri: ClassVar[Dict[str, str]] = {} 

31 

32 def __init__(self) -> None: 

33 """ 

34 Constructor of the ``AbstractSet`` class. 

35 """ 

36 self.res_to_entity: Dict[str, E] = {} 

37 

38 def graphs(self) -> List[TripleLite]: 

39 """ 

40 A utility method that allows to retrieve the list of ``TripleLite`` 

41 instances corresponding to each entity contained in the set. 

42 

43 :return: The requested list of graphs 

44 """ 

45 result: List[TripleLite] = [] 

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

47 if len(entity.g) > 0: 

48 result.append(entity.g) 

49 return result 

50 

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

52 """ 

53 Support for pickle serialization. 

54 

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

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

57 pickle protocol support for AbstractSet instances. 

58 """ 

59 return vars(self).copy() 

60 

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

62 """ 

63 Support for pickle deserialization. 

64 

65 Restores the AbstractSet state from a pickled representation. 

66 """ 

67 vars(self).update(state) 

68 

69 @abstractmethod 

70 def get_entity(self, res: str) -> Optional[E]: 

71 """ 

72 Method signature for concrete implementations that allow 

73 to retrieve a contained entity identified by its URI. 

74 

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

76 :type res: URIRef 

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

78 """ 

79 raise NotImplementedError 

80 

81 @staticmethod 

82 def get_graph_iri(g: TripleLite) -> str: 

83 """ 

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

85 the name of a given named graph. 

86 

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

88 

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

90 :type g: Graph 

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

92 """ 

93 return str(g.identifier)