Coverage for oc_ocdm / prov / prov_entity.py: 95%

1#!/usr/bin/python 

2 

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

4# 

5# SPDX-License-Identifier: ISC 

6 

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

8from __future__ import annotations 

9 

10from typing import TYPE_CHECKING 

11 

12from triplelite import TripleLite 

13 

14from oc_ocdm.abstract_entity import AbstractEntity 

15from oc_ocdm.constants import Namespace 

16from oc_ocdm.graph.graph_entity import GraphEntity 

17 

18if TYPE_CHECKING: 

19 from typing import ClassVar, Dict, Optional 

20 

21 from oc_ocdm.prov.prov_set import ProvSet 

22 

23 

24class ProvEntity(AbstractEntity): 

25 """Snapshot of entity metadata (short: se): a particular snapshot recording the 

26 metadata associated with an individual entity (either a bibliographic entity or an 

27 identifier) at a particular date and time, including the agent, such as a person, 

28 organisation or automated process that created or modified the entity metadata. 

29 """ 

30 

31 PROV: ClassVar[Namespace] = Namespace("http://www.w3.org/ns/prov#") 

32 

33 iri_entity: ClassVar[str] = PROV.Entity 

34 iri_generated_at_time: ClassVar[str] = PROV.generatedAtTime 

35 iri_invalidated_at_time: ClassVar[str] = PROV.invalidatedAtTime 

36 iri_specialization_of: ClassVar[str] = PROV.specializationOf 

37 iri_was_derived_from: ClassVar[str] = PROV.wasDerivedFrom 

38 iri_had_primary_source: ClassVar[str] = PROV.hadPrimarySource 

39 iri_was_attributed_to: ClassVar[str] = PROV.wasAttributedTo 

40 iri_description: ClassVar[str] = GraphEntity.DCTERMS.description 

41 iri_has_update_query: ClassVar[str] = GraphEntity.OCO.hasUpdateQuery 

42 

43 short_name_to_type_iri: ClassVar[Dict[str, str]] = {"se": iri_entity} 

44 

45 def __init__( 

46 self, 

47 prov_subject: GraphEntity, 

48 g: TripleLite, 

49 p_set: ProvSet, 

50 res: Optional[str] = None, 

51 resp_agent: Optional[str] = None, 

52 source: Optional[str] = None, 

53 count: Optional[str] = None, 

54 label: Optional[str] = None, 

55 short_name: str = "se", 

56 ) -> None: 

57 super(ProvEntity, self).__init__() 

58 self.prov_subject: GraphEntity = prov_subject 

59 

60 self.g: TripleLite = g 

61 self.resp_agent: Optional[str] = resp_agent 

62 self.source: Optional[str] = source 

63 self.short_name: str = short_name 

64 self.p_set: ProvSet = p_set 

65 

66 if res is not None and count is not None: 

67 raise ValueError("'res' and 'count' are mutually exclusive: provide one or the other") 

68 if res is not None: 

69 self.res = res 

70 elif count is not None: 

71 self.res = self._generate_new_res(g, count, short_name) 

72 else: 

73 raise ValueError("Either 'res' or 'count' must be provided") 

74 

75 if self.res not in p_set.res_to_entity: 

76 p_set.res_to_entity[self.res] = self 

77 

78 self._create_type(self.short_name_to_type_iri[short_name]) 

79 if label is not None: 

80 self.create_label(label) 

81 

82 @staticmethod 

83 def _generate_new_res(g: TripleLite, count: str, short_name: str) -> str: 

84 return str(g.identifier) + short_name + "/" + count