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 rdflib import Graph, Namespace, URIRef 

13 

14from oc_ocdm.abstract_entity import AbstractEntity 

15from oc_ocdm.graph.graph_entity import GraphEntity 

16 

17if TYPE_CHECKING: 

18 from typing import ClassVar, Dict, Optional 

19 

20 from oc_ocdm.prov.prov_set import ProvSet 

21 

22 

23class ProvEntity(AbstractEntity): 

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

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

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

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

28 """ 

29 

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

31 

32 iri_entity: ClassVar[URIRef] = PROV.Entity 

33 iri_generated_at_time: ClassVar[URIRef] = PROV.generatedAtTime 

34 iri_invalidated_at_time: ClassVar[URIRef] = PROV.invalidatedAtTime 

35 iri_specialization_of: ClassVar[URIRef] = PROV.specializationOf 

36 iri_was_derived_from: ClassVar[URIRef] = PROV.wasDerivedFrom 

37 iri_had_primary_source: ClassVar[URIRef] = PROV.hadPrimarySource 

38 iri_was_attributed_to: ClassVar[URIRef] = PROV.wasAttributedTo 

39 iri_description: ClassVar[URIRef] = GraphEntity.DCTERMS.description 

40 iri_has_update_query: ClassVar[URIRef] = GraphEntity.OCO.hasUpdateQuery 

41 

42 short_name_to_type_iri: ClassVar[Dict[str, URIRef]] = { 

43 'se': iri_entity 

44 } 

45 

46 def __init__(self, prov_subject: GraphEntity, g: Graph, p_set: ProvSet, 

47 res: Optional[URIRef] = None, resp_agent: Optional[str] = None, source: Optional[str] = None, 

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

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

50 super(ProvEntity, self).__init__() 

51 self.prov_subject: GraphEntity = prov_subject 

52 

53 self.g: Graph = g 

54 self.resp_agent: Optional[str] = resp_agent 

55 self.source: Optional[str] = source 

56 self.short_name: str = short_name 

57 self.p_set: ProvSet = p_set 

58 

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

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

61 if res is not None: 

62 self.res = res 

63 elif count is not None: 

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

65 else: 

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

67 

68 if p_set is not None: 

69 if self.res not in p_set.res_to_entity: 

70 p_set.res_to_entity[self.res] = self 

71 

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

73 if label is not None: 

74 self.create_label(label) 

75 

76 @staticmethod 

77 def _generate_new_res(g: Graph, count: str, short_name: str) -> URIRef: 

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