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]] = { 

44 'se': iri_entity 

45 } 

46 

47 def __init__(self, prov_subject: GraphEntity, g: TripleLite, p_set: ProvSet, 

48 res: Optional[str] = None, resp_agent: Optional[str] = None, source: Optional[str] = None, 

49 count: Optional[str] = None, label: Optional[str] = None, 

50 short_name: str = "se") -> None: 

51 super(ProvEntity, self).__init__() 

52 self.prov_subject: GraphEntity = prov_subject 

53 

54 self.g: TripleLite = g 

55 self.resp_agent: Optional[str] = resp_agent 

56 self.source: Optional[str] = source 

57 self.short_name: str = short_name 

58 self.p_set: ProvSet = p_set 

59 

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

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

62 if res is not None: 

63 self.res = res 

64 elif count is not None: 

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

66 else: 

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

68 

69 if p_set is not None: 

70 if self.res not in p_set.res_to_entity: 

71 p_set.res_to_entity[self.res] = self 

72 

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

74 if label is not None: 

75 self.create_label(label) 

76 

77 @staticmethod 

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

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