Coverage for oc_ocdm / graph / graph_entity.py: 97%
257 statements
« prev ^ index » next coverage.py v7.13.4, created at 2026-07-06 20:05 +0000
1# SPDX-FileCopyrightText: 2020-2022 Simone Persiani <iosonopersia@gmail.com>
2# SPDX-FileCopyrightText: 2022-2024 Arcangelo Massari <arcangelo.massari@unibo.it>
3# SPDX-FileCopyrightText: 2024 martasoricetti <marta.soricetti@studio.unibo.it>
4#
5# SPDX-License-Identifier: ISC
7from __future__ import annotations
9from typing import TYPE_CHECKING
11from triplelite import RDFTerm, SubgraphView, Triple, TripleLite
13from oc_ocdm.abstract_entity import AbstractEntity
14from oc_ocdm.constants import RDF_TYPE, Namespace
16if TYPE_CHECKING:
17 from typing import ClassVar, Dict, List, Optional
19 from oc_ocdm.graph.graph_set import GraphSet
22class GraphEntity(AbstractEntity):
23 BIRO: ClassVar[Namespace] = Namespace("http://purl.org/spar/biro/")
24 C4O: ClassVar[Namespace] = Namespace("http://purl.org/spar/c4o/")
25 CO: ClassVar[Namespace] = Namespace("http://purl.org/co/")
26 CITO: ClassVar[Namespace] = Namespace("http://purl.org/spar/cito/")
27 DATACITE: ClassVar[Namespace] = Namespace("http://purl.org/spar/datacite/")
28 DCTERMS: ClassVar[Namespace] = Namespace("http://purl.org/dc/terms/")
29 DEO: ClassVar[Namespace] = Namespace("http://purl.org/spar/deo/")
30 DOCO: ClassVar[Namespace] = Namespace("http://purl.org/spar/doco/")
31 FABIO: ClassVar[Namespace] = Namespace("http://purl.org/spar/fabio/")
32 FOAF: ClassVar[Namespace] = Namespace("http://xmlns.com/foaf/0.1/")
33 FR: ClassVar[Namespace] = Namespace("http://purl.org/spar/fr/")
34 FRBR: ClassVar[Namespace] = Namespace("http://purl.org/vocab/frbr/core#")
35 LITERAL: ClassVar[Namespace] = Namespace("http://www.essepuntato.it/2010/06/literalreification/")
36 OA: ClassVar[Namespace] = Namespace("http://www.w3.org/ns/oa#")
37 OCO: ClassVar[Namespace] = Namespace("https://w3id.org/oc/ontology/")
38 PRISM: ClassVar[Namespace] = Namespace("http://prismstandard.org/namespaces/basic/2.0/")
39 PRO: ClassVar[Namespace] = Namespace("http://purl.org/spar/pro/")
41 iri_has_subtitle: ClassVar[str] = FABIO.hasSubtitle
42 iri_has_publication_date: ClassVar[str] = PRISM.publicationDate
43 iri_bibliographic_reference: ClassVar[str] = BIRO.BibliographicReference
44 iri_references: ClassVar[str] = BIRO.references
45 iri_denotes: ClassVar[str] = C4O.denotes
46 iri_has_content: ClassVar[str] = C4O.hasContent
47 iri_intextref_pointer: ClassVar[str] = C4O.InTextReferencePointer
48 iri_is_context_of: ClassVar[str] = C4O.isContextOf
49 iri_singleloc_pointer_list: ClassVar[str] = C4O.SingleLocationPointerList
50 iri_has_element: ClassVar[str] = CO.element
51 iri_citation: ClassVar[str] = CITO.Citation
52 iri_cites: ClassVar[str] = CITO.cites
53 iri_citation_characterisation: ClassVar[str] = CITO.hasCitationCharacterisation
54 iri_has_citing_entity: ClassVar[str] = CITO.hasCitingEntity
55 iri_has_cited_entity: ClassVar[str] = CITO.hasCitedEntity
56 iri_openalex: ClassVar[str] = DATACITE.openalex
57 iri_arxiv: ClassVar[str] = DATACITE.arxiv
58 iri_oci: ClassVar[str] = DATACITE.oci
59 iri_doi: ClassVar[str] = DATACITE.doi
60 iri_pmid: ClassVar[str] = DATACITE.pmid
61 iri_pmcid: ClassVar[str] = DATACITE.pmcid
62 iri_orcid: ClassVar[str] = DATACITE.orcid
63 iri_xpath: ClassVar[str] = DATACITE["local-resource-identifier-scheme"]
64 iri_intrepid: ClassVar[str] = DATACITE.intrepid
65 iri_xmlid: ClassVar[str] = DATACITE["local-resource-identifier-scheme"]
66 iri_has_identifier: ClassVar[str] = DATACITE.hasIdentifier
67 iri_identifier: ClassVar[str] = DATACITE.Identifier
68 iri_isbn: ClassVar[str] = DATACITE.isbn
69 iri_issn: ClassVar[str] = DATACITE.issn
70 iri_url: ClassVar[str] = DATACITE.url
71 iri_uses_identifier_scheme: ClassVar[str] = DATACITE.usesIdentifierScheme
72 iri_title: ClassVar[str] = DCTERMS.title
73 iri_caption: ClassVar[str] = DEO.Caption
74 iri_discourse_element: ClassVar[str] = DEO.DiscourseElement
75 iri_footnote: ClassVar[str] = DOCO.Footnote
76 iri_paragraph: ClassVar[str] = DOCO.Paragraph
77 iri_part: ClassVar[str] = DOCO.Part
78 iri_section: ClassVar[str] = DOCO.Section
79 iri_introduction: ClassVar[str] = DEO.Introduction
80 iri_methods: ClassVar[str] = DEO.Methods
81 iri_materials: ClassVar[str] = DEO.Materials
82 iri_related_work: ClassVar[str] = DEO.RelatedWork
83 iri_results: ClassVar[str] = DEO.Results
84 iri_discussion: ClassVar[str] = DEO.Discussion
85 iri_conclusion: ClassVar[str] = DEO.Conclusion
86 iri_section_title: ClassVar[str] = DOCO.SectionTitle
87 iri_sentence: ClassVar[str] = DOCO.Sentence
88 iri_table: ClassVar[str] = DOCO.Table
89 iri_text_chunk: ClassVar[str] = DOCO.TextChunk
90 iri_abstract: ClassVar[str] = DOCO.Abstract
91 iri_academic_proceedings: ClassVar[str] = FABIO.AcademicProceedings
92 iri_audio_document: ClassVar[str] = FABIO.AudioDocument
93 iri_book: ClassVar[str] = FABIO.Book
94 iri_book_chapter: ClassVar[str] = FABIO.BookChapter
95 iri_book_series: ClassVar[str] = FABIO.BookSeries
96 iri_book_set: ClassVar[str] = FABIO.BookSet
97 iri_computer_program: ClassVar[str] = FABIO.ComputerProgram
98 iri_data_file: ClassVar[str] = FABIO.DataFile
99 iri_data_management_plan: ClassVar[str] = FABIO.DataManagementPlan
100 iri_editorial: ClassVar[str] = FABIO.Editorial
101 iri_expression: ClassVar[str] = FABIO.Expression
102 iri_expression_collection: ClassVar[str] = FABIO.ExpressionCollection
103 iri_has_sequence_identifier: ClassVar[str] = FABIO.hasSequenceIdentifier
104 iri_journal: ClassVar[str] = FABIO.Journal
105 iri_journal_article: ClassVar[str] = FABIO.JournalArticle
106 iri_journal_editorial: ClassVar[str] = FABIO.JournalEditorial
107 iri_journal_issue: ClassVar[str] = FABIO.JournalIssue
108 iri_journal_volume: ClassVar[str] = FABIO.JournalVolume
109 iri_manifestation: ClassVar[str] = FABIO.Manifestation
110 iri_newspaper: ClassVar[str] = FABIO.Newspaper
111 iri_newspaper_article: ClassVar[str] = FABIO.NewspaperArticle
112 iri_newspaper_editorial: ClassVar[str] = FABIO.NewspaperEditorial
113 iri_newspaper_issue: ClassVar[str] = FABIO.NewspaperIssue
114 iri_peer_review: ClassVar[str] = FR.ReviewVersion
115 iri_preprint: ClassVar[str] = FABIO.Preprint
116 iri_presentation: ClassVar[str] = FABIO.Presentation
117 iri_proceedings_paper: ClassVar[str] = FABIO.ProceedingsPaper
118 iri_proceedings_series: ClassVar[str] = FABIO.Series
119 iri_reference_book: ClassVar[str] = FABIO.ReferenceBook
120 iri_reference_entry: ClassVar[str] = FABIO.ReferenceEntry
121 iri_report_document: ClassVar[str] = FABIO.ReportDocument
122 iri_retraction_notice: ClassVar[str] = FABIO.RetractionNotice
123 iri_series: ClassVar[str] = FABIO.Series
124 iri_specification_document: ClassVar[str] = FABIO.SpecificationDocument
125 iri_thesis: ClassVar[str] = FABIO.Thesis
126 iri_web_content: ClassVar[str] = FABIO.WebContent
127 iri_agent: ClassVar[str] = FOAF.Agent
128 iri_family_name: ClassVar[str] = FOAF.familyName
129 iri_given_name: ClassVar[str] = FOAF.givenName
130 iri_name: ClassVar[str] = FOAF.name
131 iri_embodiment: ClassVar[str] = FRBR.embodiment
132 iri_part_of: ClassVar[str] = FRBR.partOf
133 iri_contains_reference: ClassVar[str] = FRBR.part
134 iri_contains_de: ClassVar[str] = FRBR.part
135 iri_has_literal_value: ClassVar[str] = LITERAL.hasLiteralValue
136 iri_ending_page: ClassVar[str] = PRISM.endingPage
137 iri_starting_page: ClassVar[str] = PRISM.startingPage
138 iri_author: ClassVar[str] = PRO.author
139 iri_editor: ClassVar[str] = PRO.editor
140 iri_is_held_by: ClassVar[str] = PRO.isHeldBy
141 iri_publisher: ClassVar[str] = PRO.publisher
142 iri_is_document_context_for: ClassVar[str] = PRO.isDocumentContextFor
143 iri_role_in_time: ClassVar[str] = PRO.RoleInTime
144 iri_with_role: ClassVar[str] = PRO.withRole
145 iri_note: ClassVar[str] = OA.Annotation
146 iri_has_body: ClassVar[str] = OA.hasBody
147 iri_has_annotation: ClassVar[str] = OCO.hasAnnotation # inverse of OA.hasTarget
148 iri_has_next: ClassVar[str] = OCO.hasNext
149 iri_archival_document: ClassVar[str] = FABIO.ArchivalDocument
150 iri_viaf: ClassVar[str] = DATACITE.viaf
151 iri_crossref: ClassVar[str] = DATACITE.crossref # TODO: add to datacite!
152 iri_datacite: ClassVar[str] = DATACITE.datacite # TODO: add to datacite!
153 iri_jid: ClassVar[str] = DATACITE.jid # TODO: add to datacite!
154 iri_wikidata: ClassVar[str] = DATACITE.wikidata # TODO: add to datacite!
155 iri_wikipedia: ClassVar[str] = DATACITE.wikipedia # TODO: add to datacite!
156 iri_has_edition: ClassVar[str] = PRISM.edition
157 iri_relation: ClassVar[str] = DCTERMS.relation
158 iri_has_citation_creation_date: ClassVar[str] = CITO.hasCitationCreationDate
159 iri_has_citation_time_span: ClassVar[str] = CITO.hasCitationTimeSpan
160 iri_digital_manifestation: ClassVar[str] = FABIO.DigitalManifestation
161 iri_print_object: ClassVar[str] = FABIO.PrintObject
162 iri_has_url: ClassVar[str] = FRBR.exemplar
163 iri_self_citation: ClassVar[str] = CITO.SelfCitation
164 iri_affiliation_self_citation: ClassVar[str] = CITO.AffiliationSelfCitation
165 iri_author_network_self_citation: ClassVar[str] = CITO.AuthorNetworkSelfCitation
166 iri_author_self_citation: ClassVar[str] = CITO.AuthorSelfCitation
167 iri_funder_self_citation: ClassVar[str] = CITO.FunderSelfCitation
168 iri_journal_self_citation: ClassVar[str] = CITO.JournalSelfCitation
169 iri_journal_cartel_citation: ClassVar[str] = CITO.JournalCartelCitation
170 iri_distant_citation: ClassVar[str] = CITO.DistantCitation
171 iri_has_format: ClassVar[str] = DCTERMS.format
173 short_name_to_type_iri: ClassVar[Dict[str, str]] = {
174 "an": iri_note,
175 "ar": iri_role_in_time,
176 "be": iri_bibliographic_reference,
177 "br": iri_expression,
178 "ci": iri_citation,
179 "de": iri_discourse_element,
180 "id": iri_identifier,
181 "pl": iri_singleloc_pointer_list,
182 "ra": iri_agent,
183 "re": iri_manifestation,
184 "rp": iri_intextref_pointer,
185 }
187 def __init__(
188 self,
189 g: TripleLite,
190 g_set: GraphSet,
191 res_type: str,
192 res: str | None = None,
193 resp_agent: str | None = None,
194 source: str | None = None,
195 count: str | None = None,
196 label: str | None = None,
197 short_name: str = "",
198 preexisting_graph: SubgraphView | None = None,
199 ) -> None:
200 super(GraphEntity, self).__init__()
201 self.g: TripleLite = g
202 self.resp_agent: str | None = resp_agent
203 self.source: str | None = source
204 self.short_name: str = short_name
205 self.g_set: GraphSet = g_set
206 self._preexisting_triples: frozenset[Triple] = frozenset()
207 self._merge_list: tuple[GraphEntity, ...] = ()
208 # FLAGS
209 self._to_be_deleted: bool = False
210 self._was_merged: bool = False
211 self._is_restored: bool = False
213 # If res was not specified, create from scratch the URI reference for this entity,
214 # otherwise use the provided one
215 if res is None:
216 self.res = self._generate_new_res(g, count)
217 else:
218 self.res = res
220 # If not already done, register this GraphEntity instance inside the GraphSet
221 if self.res not in g_set.res_to_entity:
222 g_set.res_to_entity[self.res] = self
224 if preexisting_graph is not None:
225 # Triples inside self.g are entirely replaced by triples from preexisting_graph.
226 # This has maximum priority with respect to every other self.g initializations.
227 # It's fundamental that the preexisting graph gets passed as an argument of the constructor:
228 # allowing the user to set this value later through a method would mean that the user could
229 # set the preexisting graph AFTER having modified self.g (which would not make sense).
230 self.remove_every_triple()
231 self._preexisting_triples = frozenset(
232 (self.res, p, o) for p, o in preexisting_graph.predicate_objects(self.res)
233 )
234 self.g.add_many(self._preexisting_triples)
235 else:
236 # Add mandatory information to the entity graph
237 self._create_type(res_type)
238 if label is not None:
239 self.create_label(label)
241 @staticmethod
242 def _generate_new_res(g: TripleLite, count: str | None) -> str:
243 assert count is not None
244 return str(g.identifier) + count
246 @property
247 def to_be_deleted(self) -> bool:
248 return self._to_be_deleted
250 @property
251 def was_merged(self) -> bool:
252 return self._was_merged
254 @property
255 def merge_list(self) -> tuple[GraphEntity, ...]:
256 return self._merge_list
258 @property
259 def preexisting_triples(self) -> frozenset[Triple]:
260 return self._preexisting_triples
262 @property
263 def is_restored(self) -> bool:
264 """Indicates if this entity was restored after being deleted."""
265 return self._is_restored
267 def mark_as_restored(self) -> None:
268 """
269 Marks an entity as being restored after deletion.
271 This state signals to the provenance system that:
272 - No new invalidation time should be generated for the previous snapshot
273 - The original deletion snapshot's invalidation time should be preserved
274 - The entity should be treated as restored rather than newly created
275 """
276 self._to_be_deleted = False
277 self._is_restored = True
279 def mark_as_to_be_deleted(self) -> None:
280 # Here we must REMOVE triples pointing
281 # to 'self' [THIS CANNOT BE UNDONE]:
282 for res, entity in self.g_set.res_to_entity.items():
283 triples_list: List[Triple] = list(entity.g.triples((res, None, RDFTerm("uri", str(self.res)))))
284 for triple in triples_list:
285 entity.g.remove(triple)
287 self._to_be_deleted = True
289 def _get_specific_type(self) -> Optional[str]:
290 base_type_str = self.short_name_to_type_iri[self.short_name]
291 for _, _, type_uri in self.g.triples((self.res, RDF_TYPE, None)):
292 if type_uri.type == "uri" and type_uri.value != base_type_str:
293 return type_uri.value
294 return None
296 def merge(self, other: object, prefer_self: bool = False) -> None:
297 """
298 **WARNING:** ``GraphEntity`` **is an abstract class that cannot be instantiated at runtime.
299 As such, it's only possible to execute this method on entities generated from**
300 ``GraphEntity``'s **subclasses. Please, refer to their documentation of the** `merge` **method.**
302 :param other: The entity which will be marked as to be deleted and whose properties will
303 be merged into the current entity.
304 :type other: GraphEntity
305 :param prefer_self: If True, keep existing functional values from the current entity
306 :type prefer_self: bool
307 :raises TypeError: if the parameter is not of the same entity type
308 :return: None
309 """
310 if not isinstance(other, GraphEntity) or other.short_name != self.short_name:
311 raise TypeError(
312 f"[{self.__class__.__name__}.merge] Expected entity type: {self.short_name}. "
313 f"Provided: {type(other).__name__}."
314 )
316 # Redirect triples pointing to 'other' to point to 'self'
317 for res, entity in self.g_set.res_to_entity.items():
318 triples_list: List[Triple] = list(entity.g.triples((res, None, RDFTerm("uri", str(other.res)))))
319 for triple in triples_list:
320 entity.g.remove(triple)
321 new_triple = (triple[0], triple[1], RDFTerm("uri", str(self.res)))
322 entity.g.add(new_triple)
324 self_specific_type = self._get_specific_type()
325 other_specific_type = other._get_specific_type()
327 final_specific_type = None
328 if prefer_self and self_specific_type:
329 final_specific_type = self_specific_type
330 elif other_specific_type:
331 final_specific_type = other_specific_type
332 elif self_specific_type:
333 final_specific_type = self_specific_type
335 self.g.remove((self.res, RDF_TYPE, None))
336 base_type = self.short_name_to_type_iri[self.short_name]
337 self.g.add((self.res, RDF_TYPE, RDFTerm("uri", base_type)))
338 if final_specific_type:
339 self.g.add((self.res, RDF_TYPE, RDFTerm("uri", final_specific_type)))
341 label: Optional[str] = other.get_label()
342 if label is not None and self._should_merge_functional_property(self.get_label(), prefer_self):
343 self.create_label(label)
345 self._was_merged = True
346 self._merge_list = (*self._merge_list, other)
348 # 'other' must be deleted AFTER the redirection of
349 # triples pointing to it, since mark_as_to_be_deleted
350 # also removes every triple pointing to 'other'
351 other.mark_as_to_be_deleted()
353 self._merge_properties(other, prefer_self)
355 def _merge_properties(self, other: GraphEntity, prefer_self: bool) -> None:
356 pass
358 @staticmethod
359 def _should_merge_functional_property(current_value: object | None, prefer_self: bool) -> bool:
360 return not prefer_self or current_value is None
362 def commit_changes(self) -> None:
363 if self._to_be_deleted:
364 self._preexisting_triples = frozenset()
365 self.remove_every_triple()
366 else:
367 self._preexisting_triples = frozenset(self.g.triples((self.res, None, None)))
368 self._is_restored = False
369 self._to_be_deleted = False
370 self._was_merged = False
371 self._merge_list = tuple()